use anyhow::Result;
use chrono::{DateTime, Utc};
use std::borrow::Cow;
use std::path::{Path, PathBuf};
pub fn canonicalize_simplified(path: impl AsRef<Path>) -> std::io::Result<PathBuf> {
std::fs::canonicalize(path).map(simplify_verbatim)
}
pub fn simplify_verbatim(path: PathBuf) -> PathBuf {
let simplified = match path.to_str() {
Some(text) => match simplify_verbatim_str(text) {
Cow::Borrowed(unchanged) if unchanged.len() == text.len() => None,
simplified => Some(simplified.into_owned()),
},
None => None,
};
match simplified {
Some(text) => PathBuf::from(text),
None => path,
}
}
fn simplify_verbatim_str(text: &str) -> Cow<'_, str> {
const MAX_PATH: usize = 260;
if let Some(share) = text.strip_prefix(r"\\?\UNC\") {
let plain = format!(r"\\{share}");
if plain.len() < MAX_PATH {
return Cow::Owned(plain);
}
return Cow::Borrowed(text);
}
if let Some(rest) = text.strip_prefix(r"\\?\") {
let mut head = rest.chars();
let drive = matches!(
(head.next(), head.next(), head.next()),
(Some(letter), Some(':'), Some('\\')) if letter.is_ascii_alphabetic()
);
if drive && rest.len() < MAX_PATH {
return Cow::Borrowed(rest);
}
}
Cow::Borrowed(text)
}
pub fn relfn2path(uri: &str, docname: &str, srcdir: &Path) -> PathBuf {
let mut path = srcdir.to_path_buf();
for segment in relfn2path_rel(uri, docname).split('/') {
if !segment.is_empty() {
path.push(segment);
}
}
path
}
pub fn relfn2path_io(uri: &str, docname: &str, srcdir: &Path) -> PathBuf {
let mut path = srcdir.to_path_buf();
for segment in relfn2path_join(uri, docname).split('/') {
if !segment.is_empty() {
path.push(segment);
}
}
resolve_path(&path)
}
fn relfn2path_join(uri: &str, docname: &str) -> String {
match uri.strip_prefix('/') {
Some(rooted) => rooted.to_string(),
None => match docname.rsplit_once('/') {
Some((dir, _)) => format!("{dir}/{uri}"),
None => uri.to_string(),
},
}
}
pub(crate) fn resolve_path(path: &Path) -> PathBuf {
use std::path::Component;
let mut resolved = PathBuf::new();
for component in path.components() {
match component {
Component::Prefix(_) | Component::RootDir => resolved.push(component),
Component::CurDir => {}
Component::ParentDir => {
resolved.pop();
}
Component::Normal(name) => {
resolved.push(name);
if let Ok(real) = canonicalize_simplified(&resolved) {
resolved = real;
}
}
}
}
resolved
}
pub fn relfn2path_rel(uri: &str, docname: &str) -> String {
let relative = match uri.strip_prefix('/') {
Some(rooted) => rooted.to_string(),
None => match docname.rsplit_once('/') {
Some((dir, _)) => format!("{dir}/{uri}"),
None => uri.to_string(),
},
};
normalize_dot_segments(&relative)
}
pub fn normalize_dot_segments(relative: &str) -> String {
let mut segments: Vec<&str> = Vec::new();
for segment in relative.split('/') {
match segment {
"" | "." => {}
".." => {
if matches!(segments.last(), Some(&last) if last != "..") {
segments.pop();
} else {
segments.push("..");
}
}
other => segments.push(other),
}
}
segments.join("/")
}
pub(crate) fn relative_path_walk_up(path: &Path, root: &Path) -> String {
use std::path::Component;
let path_parts: Vec<Component<'_>> = path.components().collect();
let root_parts: Vec<Component<'_>> = root.components().collect();
let anchors_differ = match (path_parts.first(), root_parts.first()) {
(Some(Component::Prefix(a)), Some(Component::Prefix(b))) => a != b,
(Some(Component::Prefix(_)), _) | (_, Some(Component::Prefix(_))) => true,
_ => false,
};
if anchors_differ {
return path.to_string_lossy().replace('\\', "/");
}
let common = path_parts
.iter()
.zip(root_parts.iter())
.take_while(|(a, b)| a == b)
.count();
let mut segments: Vec<String> = vec!["..".to_string(); root_parts.len() - common];
segments.extend(
path_parts[common..]
.iter()
.map(|c| c.as_os_str().to_string_lossy().into_owned()),
);
segments.join("/")
}
pub(crate) fn py_isspace(c: char) -> bool {
c.is_whitespace() || matches!(c, '\x1c'..='\x1f')
}
pub(crate) fn py_repr_str(s: &str) -> String {
let quote = if s.contains('\'') && !s.contains('"') {
'"'
} else {
'\''
};
let mut out = String::with_capacity(s.len() + 2);
out.push(quote);
for c in s.chars() {
match c {
'\\' => out.push_str("\\\\"),
'\n' => out.push_str("\\n"),
'\r' => out.push_str("\\r"),
'\t' => out.push_str("\\t"),
c if c == quote => {
out.push('\\');
out.push(c);
}
c if c != ' ' && (c.is_control() || c.is_whitespace()) => {
let n = c as u32;
if n <= 0xff {
out.push_str(&format!("\\x{n:02x}"));
} else if n <= 0xffff {
out.push_str(&format!("\\u{n:04x}"));
} else {
out.push_str(&format!("\\U{n:08x}"));
}
}
c => out.push(c),
}
}
out.push(quote);
out
}
pub(crate) fn py_split(s: &str) -> impl Iterator<Item = &str> {
s.split(py_isspace).filter(|w| !w.is_empty())
}
pub fn path2doc(path: &Path, srcdir: &Path) -> Option<String> {
let rel = path.strip_prefix(srcdir).ok()?;
let rel = rel.to_str()?.replace('\\', "/");
Some(rel.strip_suffix(".rst")?.to_string())
}
#[derive(Debug)]
pub struct ProjectStats {
pub source_files: usize,
pub total_lines: usize,
pub avg_file_size_kb: f64,
pub largest_file_kb: f64,
pub max_depth: usize,
pub cross_references: usize,
}
pub async fn analyze_project(source_dir: &Path) -> Result<ProjectStats> {
let mut state = AnalysisState {
source_files: 0,
total_lines: 0,
total_size_bytes: 0,
largest_file_kb: 0.0,
max_depth: 0,
cross_references: 0,
};
analyze_directory_sync(source_dir, source_dir, 0, &mut state)?;
let avg_file_size_kb = if state.source_files > 0 {
(state.total_size_bytes as f64) / (state.source_files as f64) / 1024.0
} else {
0.0
};
Ok(ProjectStats {
source_files: state.source_files,
total_lines: state.total_lines,
avg_file_size_kb,
largest_file_kb: state.largest_file_kb,
max_depth: state.max_depth,
cross_references: state.cross_references,
})
}
struct AnalysisState {
source_files: usize,
total_lines: usize,
total_size_bytes: u64,
largest_file_kb: f64,
max_depth: usize,
cross_references: usize,
}
fn analyze_directory_sync(
dir: &Path,
_root_dir: &Path,
current_depth: usize,
state: &mut AnalysisState,
) -> Result<()> {
state.max_depth = state.max_depth.max(current_depth);
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
if let Some(name) = path.file_name() {
if name.to_string_lossy().starts_with('.') {
continue;
}
}
analyze_directory_sync(&path, _root_dir, current_depth + 1, state)?;
} else if is_source_file(&path) {
state.source_files += 1;
let metadata = std::fs::metadata(&path)?;
let file_size_bytes = metadata.len();
let file_size_kb = file_size_bytes as f64 / 1024.0;
state.total_size_bytes += file_size_bytes;
state.largest_file_kb = state.largest_file_kb.max(file_size_kb);
if let Ok(content) = std::fs::read_to_string(&path) {
state.total_lines += content.lines().count();
state.cross_references += count_cross_references(&content);
}
}
}
Ok(())
}
pub fn is_source_file(path: &Path) -> bool {
if let Some(ext) = path.extension() {
matches!(ext.to_string_lossy().as_ref(), "rst" | "md" | "txt")
} else {
false
}
}
pub fn count_cross_references(content: &str) -> usize {
let patterns = [
r":doc:`",
r":ref:`",
r":func:`",
r":class:`",
r":meth:`",
r":attr:`",
r":mod:`",
r":py:",
r".. _",
r"`~",
];
let mut count = 0;
for pattern in &patterns {
count += content.matches(pattern).count();
}
count
}
pub fn get_file_mtime(path: &Path) -> Result<DateTime<Utc>> {
let metadata = std::fs::metadata(path)?;
let mtime = metadata.modified()?;
Ok(DateTime::from(mtime))
}
pub async fn calculate_directory_size(dir: &Path) -> Result<u64> {
calculate_directory_size_sync(dir)
}
fn calculate_directory_size_sync(dir: &Path) -> Result<u64> {
let mut total_size = 0;
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
total_size += calculate_directory_size_sync(&path)?;
} else {
let metadata = std::fs::metadata(&path)?;
total_size += metadata.len();
}
}
Ok(total_size)
}
pub async fn copy_dir_recursive(src: &Path, dst: &Path) -> Result<()> {
copy_dir_recursive_sync(src, dst)
}
fn copy_dir_recursive_sync(src: &Path, dst: &Path) -> Result<()> {
std::fs::create_dir_all(dst)?;
for entry in std::fs::read_dir(src)? {
let entry = entry?;
let src_path = entry.path();
let dst_path = dst.join(entry.file_name());
if src_path.is_dir() {
copy_dir_recursive_sync(&src_path, &dst_path)?;
} else {
std::fs::copy(&src_path, &dst_path)?;
}
}
Ok(())
}
#[allow(dead_code)]
pub fn format_duration(duration: std::time::Duration) -> String {
let secs = duration.as_secs();
let millis = duration.subsec_millis();
if secs > 0 {
format!("{}.{:03}s", secs, millis)
} else {
format!("{}ms", millis)
}
}
#[allow(dead_code)]
pub fn format_bytes(bytes: u64) -> String {
const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"];
if bytes == 0 {
return "0 B".to_string();
}
let mut size = bytes as f64;
let mut unit_index = 0;
while size >= 1024.0 && unit_index < UNITS.len() - 1 {
size /= 1024.0;
unit_index += 1;
}
format!("{:.1} {}", size, UNITS[unit_index])
}
#[allow(dead_code)]
pub fn format_date(fmt: &str, _language: &Option<String>) -> String {
let now = chrono::Utc::now();
match fmt {
"%b %d, %Y" => now.format("%b %d, %Y").to_string(),
"%B %d, %Y" => now.format("%B %d, %Y").to_string(),
"%Y-%m-%d" => now.format("%Y-%m-%d").to_string(),
"%Y-%m-%d %H:%M:%S" => now.format("%Y-%m-%d %H:%M:%S").to_string(),
_ => {
match chrono::DateTime::parse_from_str(&now.to_rfc3339(), "%+") {
Ok(dt) => dt.format(fmt).to_string(),
Err(_) => now.format("%Y-%m-%d").to_string(),
}
}
}
}
#[allow(dead_code)]
pub async fn ensure_dir(path: &Path) -> Result<()> {
use tokio::fs;
if !path.exists() {
fs::create_dir_all(path).await?;
}
Ok(())
}
#[allow(dead_code)]
pub fn relative_uri(from: &str, to: &str, suffix: &str) -> String {
use std::path::Path;
let from_path = Path::new(from);
let to_path = Path::new(to);
if let Some(rel_path) =
pathdiff::diff_paths(to_path, from_path.parent().unwrap_or(Path::new("")))
{
let mut result = rel_path.to_string_lossy().to_string();
if !suffix.is_empty() && !result.ends_with(suffix) {
result.push_str(suffix);
}
result.replace('\\', "/") } else {
format!("{}{}", to, suffix)
}
}
#[allow(dead_code)]
pub async fn copy_dir_all(src: &Path, dst: &Path) -> Result<()> {
use tokio::fs;
ensure_dir(dst).await?;
let mut entries = fs::read_dir(src).await?;
while let Some(entry) = entries.next_entry().await? {
let entry_path = entry.path();
let file_name = entry.file_name();
let dest_path = dst.join(file_name);
if entry_path.is_dir() {
Box::pin(copy_dir_all(&entry_path, &dest_path)).await?;
} else {
if let Some(parent) = dest_path.parent() {
ensure_dir(parent).await?;
}
fs::copy(&entry_path, &dest_path).await?;
}
}
Ok(())
}
pub(crate) fn py_splitlines(text: &str) -> Vec<&str> {
let is_boundary = |c: char| {
matches!(
c,
'\n' | '\r'
| '\x0b'
| '\x0c'
| '\x1c'
| '\x1d'
| '\x1e'
| '\u{85}'
| '\u{2028}'
| '\u{2029}'
)
};
let mut out = Vec::new();
let mut start = 0usize;
let mut chars = text.char_indices().peekable();
while let Some((i, c)) = chars.next() {
if is_boundary(c) {
out.push(&text[start..i]);
if c == '\r' {
if let Some(&(_, '\n')) = chars.peek() {
chars.next();
}
}
start = chars.peek().map(|&(j, _)| j).unwrap_or(text.len());
}
}
if start < text.len() {
out.push(&text[start..]);
}
out
}
#[cfg(test)]
mod path_tests {
use super::*;
#[test]
fn relfn2path_rel_resolves_docname_relative_and_rooted_forms() {
assert_eq!(
relfn2path_rel("part.rst", "chapters/intro"),
"chapters/part.rst"
);
assert_eq!(
relfn2path_rel("/sub/abs.rst", "chapters/intro"),
"sub/abs.rst"
);
assert_eq!(
relfn2path_rel("../img/./pic.png", "chapters/intro"),
"img/pic.png"
);
assert_eq!(relfn2path_rel("x.rst", "index"), "x.rst");
assert_eq!(relfn2path_rel("../outside.rst", "index"), "../outside.rst");
}
#[test]
fn relfn2path_joins_the_rel_half_onto_srcdir() {
assert_eq!(
relfn2path("part.rst", "chapters/intro", Path::new("/src")),
PathBuf::from("/src/chapters/part.rst")
);
}
#[test]
fn path2doc_maps_rst_under_srcdir_and_nothing_else() {
let srcdir = Path::new("/src");
assert_eq!(
path2doc(Path::new("/src/part.rst"), srcdir),
Some("part".into())
);
assert_eq!(
path2doc(Path::new("/src/sub/abs_part.rst"), srcdir),
Some("sub/abs_part".into())
);
assert_eq!(path2doc(Path::new("/src/data.txt"), srcdir), None);
assert_eq!(path2doc(Path::new("/src/notes.md"), srcdir), None);
assert_eq!(path2doc(Path::new("/elsewhere/part.rst"), srcdir), None);
}
#[test]
fn relative_path_walk_up_matches_sphinxs_relative_to() {
let root = Path::new("/base/src");
assert_eq!(
relative_path_walk_up(Path::new("/base/src/a/c.rst"), root),
"a/c.rst"
);
assert_eq!(
relative_path_walk_up(Path::new("/base/ext/part.rst"), root),
"../ext/part.rst"
);
assert_eq!(
relative_path_walk_up(Path::new("/other/x.txt"), root),
"../../other/x.txt"
);
assert_eq!(relative_path_walk_up(Path::new("/base/src"), root), "");
}
#[test]
fn py_isspace_is_unicode_whitespace_plus_the_c0_separators() {
for c in [
' ', '\t', '\n', '\u{a0}', '\u{3000}', '\x1c', '\x1d', '\x1e', '\x1f',
] {
assert!(py_isspace(c), "{c:?}");
}
for c in ['a', '\x00', '\x1b', '\u{200b}'] {
assert!(!py_isspace(c), "{c:?}");
}
assert!(!'\x1f'.is_whitespace(), "the case Rust's predicate misses");
}
#[test]
fn py_repr_str_quotes_and_escapes_like_cpython() {
assert_eq!(py_repr_str("a"), "'a'");
assert_eq!(py_repr_str("it's"), "\"it's\"");
assert_eq!(py_repr_str("say \"hi\""), "'say \"hi\"'");
assert_eq!(py_repr_str("both ' and \""), "'both \\' and \"'");
assert_eq!(py_repr_str("a\\b"), "'a\\\\b'");
assert_eq!(py_repr_str("a\nb"), "'a\\nb'");
assert_eq!(py_repr_str("a\tb\rc"), "'a\\tb\\rc'");
assert_eq!(py_repr_str("term\u{a0}"), "'term\\xa0'");
assert_eq!(py_repr_str("foo\u{a0}bar"), "'foo\\xa0bar'");
assert_eq!(py_repr_str("a\u{3000}b"), "'a\\u3000b'");
assert_eq!(py_repr_str("a\u{85}b"), "'a\\x85b'");
assert_eq!(py_repr_str("a\x1fb\x7f"), "'a\\x1fb\\x7f'");
assert_eq!(py_repr_str("a\u{2028}b"), "'a\\u2028b'");
assert_eq!(py_repr_str("é ü"), "'é ü'", "printable non-ASCII stays raw");
}
#[test]
fn the_verbatim_prefix_is_stripped_back_to_the_python_spelling() {
let simplify = |text: &str| {
simplify_verbatim(PathBuf::from(text))
.to_string_lossy()
.into_owned()
};
assert_eq!(simplify(r"\\?\C:\Users\me\docs"), r"C:\Users\me\docs");
assert_eq!(simplify(r"\\?\c:\x"), r"c:\x");
assert_eq!(simplify(r"\\?\UNC\server\share\doc"), r"\\server\share\doc");
assert_eq!(simplify(r"\\?\Volume{9f8a}\x"), r"\\?\Volume{9f8a}\x");
let long = format!(r"\\?\C:\{}", "a".repeat(300));
assert_eq!(simplify(&long), long);
assert_eq!(simplify("/tmp/x/y"), "/tmp/x/y");
assert_eq!(simplify(r"C:\already\plain"), r"C:\already\plain");
assert_eq!(simplify(r"\\server\share"), r"\\server\share");
}
#[test]
fn relfn2path_io_walks_up_from_the_symlink_target() {
let base = tempfile::tempdir().unwrap();
let base = canonicalize_simplified(base.path()).unwrap();
let srcdir = base.join("src");
std::fs::create_dir_all(srcdir.join("real")).unwrap();
std::fs::create_dir_all(base.join("ext/inner")).unwrap();
std::fs::write(base.join("ext/sibling.txt"), "OUTSIDE\n").unwrap();
std::fs::write(srcdir.join("sibling.txt"), "INSIDE\n").unwrap();
#[cfg(unix)]
std::os::unix::fs::symlink(base.join("ext/inner"), srcdir.join("link")).unwrap();
assert_eq!(
relfn2path("link/../sibling.txt", "index", &srcdir),
srcdir.join("sibling.txt")
);
#[cfg(unix)]
assert_eq!(
relfn2path_io("link/../sibling.txt", "index", &srcdir),
base.join("ext/sibling.txt")
);
assert_eq!(
relfn2path_io("real/../sibling.txt", "index", &srcdir),
srcdir.join("sibling.txt")
);
assert_eq!(
relfn2path_io("real/../nothere.txt", "index", &srcdir),
srcdir.join("nothere.txt")
);
assert_eq!(
relfn2path_io("/sibling.txt", "sub/page", &srcdir),
srcdir.join("sibling.txt")
);
}
}