Files

1322 lines
53 KiB
Python

#!/usr/bin/env python3
"""PPT Master project-management CLI implementation.
Usage:
python3 scripts/project_manager.py init <project_name> [--format ppt169] [--dir <path>] [--quick-generate]
python3 scripts/project_manager.py import-sources <project_path> <source1> [<source2> ...] [--move | --copy]
python3 scripts/project_manager.py scaffold-spec <project_path>
python3 scripts/project_manager.py scaffold-lock <project_path>
python3 scripts/project_manager.py validate <project_path>
python3 scripts/project_manager.py info <project_path>
python3 scripts/project_manager.py page-context <project_path> P07 [--record-usage]
python3 scripts/project_manager.py page-context-report <project_path>
Examples:
python3 scripts/project_manager.py init demo --format ppt169
python3 scripts/project_manager.py validate projects/demo
Dependencies:
Standard library plus local PPT Master project and source-conversion modules.
"""
from __future__ import annotations
import argparse
import filecmp
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
from .page_context import (
build_page_context,
page_context_usage_report,
record_page_context_usage,
render_page_context,
)
from .paths import (
PROJECTS_ROOT,
REPO_ROOT,
SCRIPTS_DIR,
SOURCE_TO_MD_DIR,
)
from .project_specs import scaffold_project_artifact, validate_project_artifacts
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
from attribution_guard import require_skill_integrity # noqa: E402
from workflow_log import append_note # noqa: E402
try:
from project_utils import (
CANVAS_FORMATS,
get_project_info as get_project_info_common,
normalize_canvas_format,
validate_project_structure,
validate_svg_viewbox,
)
except ImportError:
tools_dir = SCRIPTS_DIR
if str(tools_dir) not in sys.path:
sys.path.insert(0, str(tools_dir))
from project_utils import ( # type: ignore
CANVAS_FORMATS,
get_project_info as get_project_info_common,
normalize_canvas_format,
validate_project_structure,
validate_svg_viewbox,
)
TOOLS_DIR = SCRIPTS_DIR
SOURCE_TO_MD_TOOLS_DIR = SOURCE_TO_MD_DIR
if str(SOURCE_TO_MD_TOOLS_DIR) not in sys.path:
sys.path.insert(0, str(SOURCE_TO_MD_TOOLS_DIR))
from _dispatcher import ( # noqa: E402
DOC_SUFFIXES,
EXCEL_SUFFIXES,
LEGACY_EXCEL_SUFFIXES,
PDF_SUFFIXES,
PRESENTATION_SUFFIXES,
build_conversion_command,
)
SOURCE_DIRNAME = "sources"
TEXT_SOURCE_SUFFIXES = {".md", ".markdown", ".txt"}
TABLE_TEXT_SUFFIXES = {".csv", ".tsv"}
BITMAP_IMAGE_SUFFIXES = {
".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif",
}
IMAGE_ASSET_SUFFIXES = BITMAP_IMAGE_SUFFIXES | {
".emf", ".wmf", ".svg",
}
def _validate_image_manifest(
payload: object,
path: Path,
) -> list[dict]:
"""Require a safe, case-insensitively unique image manifest payload."""
if not isinstance(payload, list):
raise RuntimeError(
f"Image manifest must be a JSON array: {path}"
)
seen_filenames: dict[str, str] = {}
for index, item in enumerate(payload):
if not isinstance(item, dict):
raise RuntimeError(
f"Existing image manifest item {index} must be an object: {path}"
)
filename = item.get("filename")
if (
not isinstance(filename, str)
or not filename.strip()
or filename in {".", ".."}
or "/" in filename
or "\\" in filename
or ":" in filename
or Path(filename).is_absolute()
or Path(filename).name != filename
):
raise RuntimeError(
f"Image manifest item {index} has no safe bare filename: {path}"
)
normalized_filename = filename.casefold()
if normalized_filename in seen_filenames:
raise RuntimeError(
f"Image manifest filename {filename!r} conflicts with "
f"{seen_filenames[normalized_filename]!r} (case-insensitive): {path}"
)
seen_filenames[normalized_filename] = filename
return payload
def _read_existing_image_manifest(path: Path) -> list[dict]:
"""Load an existing project image manifest or fail closed on corruption."""
if not path.exists():
return []
if not path.is_file():
raise RuntimeError(f"Existing image manifest is not a regular file: {path}")
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise RuntimeError(
f"Existing image manifest is unreadable: {path} ({exc}); "
"repair or restore it before importing more assets"
) from exc
return _validate_image_manifest(payload, path)
def _write_json_atomic(path: Path, payload: object) -> None:
"""Write JSON through a same-directory temporary file and atomic rename."""
fd, temp_name = tempfile.mkstemp(
prefix=f"{path.stem}.",
suffix=".tmp",
dir=str(path.parent),
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2)
handle.write("\n")
os.replace(temp_name, path)
except Exception:
try:
os.unlink(temp_name)
except OSError:
pass
raise
def is_url(value: str) -> bool:
"""Return whether a string looks like an HTTP(S) URL."""
parsed = urlparse(value)
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
def sanitize_name(value: str) -> str:
"""Sanitize a user-facing name into a filesystem-safe token."""
safe = "".join(ch if ch.isalnum() or ch in "-_." else "_" for ch in value.strip())
safe = safe.strip("._")
while "__" in safe:
safe = safe.replace("__", "_")
return safe[:120] or "source"
def derive_url_basename(url: str) -> str:
"""Derive a stable base filename from a URL."""
parsed = urlparse(url)
parts = [sanitize_name(parsed.netloc)]
if parsed.path and parsed.path != "/":
path_part = sanitize_name(parsed.path.strip("/").replace("/", "_"))
if path_part:
parts.append(path_part)
return "_".join(part for part in parts if part) or "web_source"
def is_within_path(path: Path, parent: Path) -> bool:
"""Return whether `path` resolves inside `parent`."""
try:
path.resolve().relative_to(parent.resolve())
return True
except ValueError:
return False
def _has_usable_import(summary: dict[str, list[str]]) -> bool:
"""Return whether import-sources produced at least one usable source artifact."""
return any(
summary.get(key)
for key in ("archived", "markdown", "assets", "images", "analysis")
)
class ProjectManager:
"""Create, inspect, validate, and populate project folders."""
CANVAS_FORMATS = CANVAS_FORMATS
def __init__(self, base_dir: str | Path | None = None) -> None:
self.base_dir = Path(base_dir) if base_dir is not None else Path.cwd() / "projects"
def scaffold_artifact(self, project_path: str, artifact: str) -> str:
"""Delegate deterministic Markdown scaffold rendering."""
return scaffold_project_artifact(Path(project_path), artifact)
def init_project(
self,
project_name: str,
canvas_format: str = "ppt169",
base_dir: str | None = None,
*,
quick_generate: bool = False,
) -> str:
base_path = Path(base_dir) if base_dir else self.base_dir
if (
not project_name
or project_name in {".", ".."}
or Path(project_name).is_absolute()
or "/" in project_name
or "\\" in project_name
):
raise ValueError(
"Project name must be a single, non-absolute path component"
)
normalized_format = normalize_canvas_format(canvas_format)
if normalized_format not in self.CANVAS_FORMATS:
available = ", ".join(sorted(self.CANVAS_FORMATS.keys()))
raise ValueError(
f"Unsupported canvas format: {canvas_format} "
f"(available: {available}; common alias: xhs -> xiaohongshu)"
)
date_str = datetime.now().strftime("%Y%m%d")
# A name already carrying a `_<format>_<YYYYMMDD>` suffix (e.g. a full
# project dir name pasted back into init) is used as-is — re-appending
# would produce `name_ppt169_20260101_ppt169_20260102`.
if re.search(rf"_{re.escape(normalized_format)}_\d{{8}}$", project_name):
project_dir_name = project_name
else:
project_dir_name = f"{project_name}_{normalized_format}_{date_str}"
project_path = base_path / project_dir_name
if not is_within_path(project_path, base_path):
raise ValueError(
f"Project directory must stay within the base directory: {base_path}"
)
if project_path.exists():
raise FileExistsError(f"Project directory already exists: {project_path}")
project_dirs = (
("svg_output",)
if quick_generate
else (
"svg_output",
"svg_final",
"images",
"icons",
"notes",
"templates",
"live_preview",
SOURCE_DIRNAME,
"analysis",
"validation",
"exports",
)
)
for rel_path in project_dirs:
(project_path / rel_path).mkdir(parents=True, exist_ok=True)
canvas_info = self.CANVAS_FORMATS[normalized_format]
if not quick_generate:
readme_path = project_path / "README.md"
readme_path.write_text(
(
f"# {project_name}\n\n"
f"- Canvas format: {normalized_format}\n"
f"- Created: {date_str}\n\n"
"## Directories\n\n"
"- `svg_output/`: raw SVG output\n"
"- `svg_final/`: self-contained SVG visual preview; may be inserted manually as an SVG image, but PowerPoint Convert to Shape is unsupported\n"
"- `images/`: runtime image pool; converter assets keep their original short filenames when possible\n"
"- `icons/`: project icon set — selected library icons copied in (via icon_sync.py) plus any custom icons you add; embedded from here at export\n"
"- `notes/`: speaker notes\n"
"- `templates/`: project templates\n"
"- `live_preview/`: browser preview runtime files and history (lock.json, server.log, edits.jsonl, annotations.jsonl)\n"
"- `sources/`: source materials and normalized markdown\n"
"- `analysis/`: machine-extracted intermediate analysis (PPTX intake, image_analysis.csv) — the pipeline's canonical must-read source/asset facts\n"
"- `validation/`: cold workflow audit log, SVG quality reports, and PPTX postflight audit reports\n"
"- `exports/`: final native DrawingML pptx deliverables only (timestamped); `_native_charts_tables.pptx` name with `--native-charts-and-tables`, `_narrated.pptx` name when narration audio is embedded\n"
"- `backup/<timestamp>/`: svg_output/ archive (always written in default-flow mode; safe to delete old timestamps)\n"
),
encoding="utf-8",
)
print(f"Project created: {project_path}")
print(f"Canvas: {canvas_info['name']} ({canvas_info['dimensions']})")
return str(project_path)
def _source_dir(self, project_path: Path) -> Path:
sources_dir = project_path / SOURCE_DIRNAME
sources_dir.mkdir(parents=True, exist_ok=True)
return sources_dir
def _analysis_dir(self, project_path: Path) -> Path:
analysis_dir = project_path / "analysis"
analysis_dir.mkdir(parents=True, exist_ok=True)
return analysis_dir
def _ensure_unique_path(self, path: Path) -> Path:
if not path.exists():
return path
suffix = path.suffix
stem = path.stem
counter = 2
while True:
candidate = path.with_name(f"{stem}_{counter}{suffix}")
if not candidate.exists():
return candidate
counter += 1
def _copy_or_move_file(self, source: Path, destination: Path, move: bool) -> Path:
try:
if source.resolve() == destination.resolve():
return destination
except FileNotFoundError:
pass
destination = self._ensure_unique_path(destination)
if move:
shutil.move(str(source), str(destination))
else:
shutil.copy2(source, destination)
return destination
def _copy_or_move_tree(self, source: Path, destination: Path, move: bool) -> Path:
try:
if source.resolve() == destination.resolve():
return destination
except FileNotFoundError:
pass
destination = self._ensure_unique_path(destination)
if move:
shutil.move(str(source), str(destination))
else:
shutil.copytree(source, destination)
return destination
def _run_tool(self, args: list[str]) -> None:
child_env = os.environ.copy()
child_env["PYTHONUTF8"] = "1"
child_env["PYTHONIOENCODING"] = "utf-8:replace"
try:
result = subprocess.run(
args,
cwd=REPO_ROOT,
check=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
env=child_env,
)
except FileNotFoundError as exc:
raise RuntimeError(f"Missing executable: {args[0]}") from exc
except subprocess.CalledProcessError as exc:
details = (exc.stderr or exc.stdout or "").strip()
raise RuntimeError(details or "tool execution failed") from exc
if result.stdout.strip():
print(result.stdout.strip())
def _import_pdf(self, pdf_path: Path, markdown_path: Path) -> None:
route = build_conversion_command(
str(pdf_path),
markdown_path,
forced_type="pdf",
)
self._run_tool(route.command)
def _import_doc(self, doc_path: Path, markdown_path: Path) -> None:
route = build_conversion_command(
str(doc_path),
markdown_path,
forced_type="doc",
)
self._run_tool(route.command)
def _import_presentation(self, presentation_path: Path, markdown_path: Path) -> None:
route = build_conversion_command(
str(presentation_path),
markdown_path,
forced_type="pptx",
)
self._run_tool(route.command)
def _import_pptx_intake(self, presentation_path: Path, project_dir: Path) -> Path:
# Multi-deck intake: each PPTX writes its own `<stem>.identity.json` /
# `<stem>.slide_library.json` and is merged into the single multi-deck
# index `analysis/source_profile.json` (one entry per source deck).
analysis_dir = self._analysis_dir(project_dir)
self._run_tool(
[
sys.executable,
str(TOOLS_DIR / "pptx_intake.py"),
str(presentation_path),
"-o",
str(analysis_dir),
]
)
return analysis_dir
def _import_excel(self, excel_path: Path, markdown_path: Path) -> None:
route = build_conversion_command(
str(excel_path),
markdown_path,
forced_type="excel",
)
self._run_tool(route.command)
def _import_url(self, url: str, markdown_path: Path) -> None:
route = build_conversion_command(
url,
markdown_path,
forced_type="web",
)
self._run_tool(route.command)
def _is_valid_imported_url_markdown(self, markdown_path: Path) -> bool:
"""Return whether web_to_md produced a usable Markdown source."""
if not markdown_path.is_file():
return False
content = markdown_path.read_text(encoding="utf-8", errors="replace")
if "[Failed URLs]:" in content:
return False
return bool(content.strip())
def _archive_url_record(self, sources_dir: Path, url: str) -> Path:
file_path = self._ensure_unique_path(sources_dir / f"{derive_url_basename(url)}.url.txt")
file_path.write_text(
f"URL: {url}\nImported: {datetime.now().isoformat(timespec='seconds')}\n",
encoding="utf-8",
)
return file_path
def _normalize_text_source(self, source_path: Path, sources_dir: Path) -> Path:
target = self._ensure_unique_path(sources_dir / f"{source_path.stem}.md")
content = source_path.read_text(encoding="utf-8", errors="replace")
target.write_text(content, encoding="utf-8")
return target
def _canonicalize_markdown_content(self, content: str) -> str:
canonical = content.replace("\r\n", "\n")
canonical = re.sub(r"(?m)^(\s*Crawled:\s+).*$", r"\1__IGNORED__", canonical)
canonical = re.sub(r"(?m)^(\s*Imported:\s+).*$", r"\1__IGNORED__", canonical)
canonical = re.sub(r"([^\s\]()/]+_files)/", "__ASSET_DIR__/", canonical)
return canonical.strip()
def _find_equivalent_markdown(self, source_path: Path, sources_dir: Path) -> Path | None:
source_content = source_path.read_text(encoding="utf-8", errors="replace")
canonical_source = self._canonicalize_markdown_content(source_content)
for existing in sorted(sources_dir.iterdir()):
if existing.suffix.lower() not in {".md", ".markdown"}:
continue
try:
if existing.resolve() == source_path.resolve():
continue
except FileNotFoundError:
pass
existing_content = existing.read_text(encoding="utf-8", errors="replace")
if self._canonicalize_markdown_content(existing_content) == canonical_source:
return existing
return None
def _companion_asset_dir(self, source_path: Path) -> Path | None:
candidate = source_path.with_name(f"{source_path.stem}_files")
if candidate.exists() and candidate.is_dir():
return candidate
return None
def _rewrite_markdown_asset_refs(
self,
markdown_path: Path,
original_asset_dirname: str,
imported_asset_dirname: str,
) -> None:
if original_asset_dirname == imported_asset_dirname:
return
content = markdown_path.read_text(encoding="utf-8", errors="replace")
updated = content.replace(f"{original_asset_dirname}/", f"{imported_asset_dirname}/")
if updated != content:
markdown_path.write_text(updated, encoding="utf-8")
def _merge_image_manifest(self, source_items: list[dict], destination_manifest: Path) -> None:
"""Merge per-source manifest items into the project-level manifest, keyed by filename."""
_validate_image_manifest(source_items, destination_manifest)
existing_data = _read_existing_image_manifest(destination_manifest)
new_by_filename: dict[str, dict] = {}
new_order: list[str] = []
for item in source_items:
filename = item.get("filename")
if not isinstance(filename, str):
continue
normalized_filename = filename.casefold()
if normalized_filename not in new_by_filename:
new_order.append(normalized_filename)
new_by_filename[normalized_filename] = item
merged: list[dict] = []
seen: set[str] = set()
for item in existing_data:
if not isinstance(item, dict):
continue
filename = item.get("filename")
if not isinstance(filename, str):
continue
normalized_filename = filename.casefold()
if normalized_filename in new_by_filename:
merged.append(new_by_filename[normalized_filename])
else:
merged.append(item)
seen.add(normalized_filename)
for normalized_filename in new_order:
if normalized_filename not in seen:
merged.append(new_by_filename[normalized_filename])
_validate_image_manifest(merged, destination_manifest)
_write_json_atomic(destination_manifest, merged)
@staticmethod
def _namespace_from_asset_dir(asset_dir: Path) -> str:
"""Derive a per-source namespace from a `<stem>_files` companion directory name."""
name = asset_dir.name
suffix = "_files"
return name[:-len(suffix)] if name.endswith(suffix) else name
def _image_destination_name(
self,
images_dir: Path,
source_file: Path,
namespace: str,
existing_manifest: dict[str, dict],
occupied_names: set[str],
) -> str:
"""Return a short unique image filename for the runtime image pool."""
candidate = images_dir / source_file.name
if candidate.name.casefold() not in occupied_names:
return source_file.name
try:
meta = existing_manifest.get(candidate.name, {})
if (
meta.get("source_namespace") == namespace
and candidate.is_file()
and filecmp.cmp(source_file, candidate, shallow=False)
):
return candidate.name
except OSError:
pass
stem = source_file.stem
suffix = source_file.suffix
counter = 2
while True:
candidate = images_dir / f"{stem}_{counter}{suffix}"
if candidate.name.casefold() not in occupied_names:
return candidate.name
try:
meta = existing_manifest.get(candidate.name, {})
if (
meta.get("source_namespace") == namespace
and candidate.is_file()
and filecmp.cmp(source_file, candidate, shallow=False)
):
return candidate.name
except OSError:
pass
counter += 1
def _propagate_image_assets(self, asset_dir: Path, project_dir: Path) -> None:
"""Copy converter-generated image assets and manifest into project images/.
Filenames are preserved when possible because source Markdown commonly
uses short names that are meaningful in context. Only real collisions
receive a compact numeric suffix.
"""
manifest_path = asset_dir / "image_manifest.json"
if not manifest_path.is_file():
return
try:
source_payload = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
print(f"[WARN] Cannot read image manifest {manifest_path}: {exc}")
return
try:
source_data = _validate_image_manifest(source_payload, manifest_path)
except RuntimeError as exc:
print(f"[WARN] {exc}")
return
images_dir = project_dir / "images"
namespace = self._namespace_from_asset_dir(asset_dir)
destination_manifest = images_dir / "image_manifest.json"
existing_data = _read_existing_image_manifest(destination_manifest)
images_dir.mkdir(parents=True, exist_ok=True)
existing_manifest = {
item["filename"]: item
for item in existing_data
}
occupied_names = {
path.name.casefold()
for path in images_dir.iterdir()
if path.is_file()
}
rename_map: dict[str, str] = {}
copied_count = 0
for source_file in sorted(asset_dir.iterdir()):
if not source_file.is_file():
continue
if source_file.suffix.lower() not in IMAGE_ASSET_SUFFIXES:
continue
new_name = self._image_destination_name(
images_dir,
source_file,
namespace,
existing_manifest,
occupied_names,
)
destination = images_dir / new_name
if source_file.resolve() != destination.resolve():
shutil.copy2(source_file, destination)
occupied_names.add(new_name.casefold())
rename_map[source_file.name] = new_name
copied_count += 1
rebased_items: list[dict] = []
for item in source_data:
if not isinstance(item, dict):
continue
original = item.get("filename")
if not isinstance(original, str):
continue
new_item = dict(item)
new_item["filename"] = rename_map.get(original, original)
new_item["source_namespace"] = namespace
rebased_items.append(new_item)
self._merge_image_manifest(rebased_items, images_dir / "image_manifest.json")
print(
f"Propagated {copied_count} image asset(s) + manifest "
f"from {asset_dir} → images/ (namespace: {namespace})"
)
def _propagate_companion_image_assets(self, markdown_path: Path, project_dir: Path) -> None:
asset_dir = markdown_path.with_name(f"{markdown_path.stem}_files")
if asset_dir.is_dir():
self._propagate_image_assets(asset_dir, project_dir)
def _import_markdown_with_assets(
self,
source_path: Path,
sources_dir: Path,
move: bool,
) -> tuple[Path, Path | None, str | None]:
archived_markdown = self._copy_or_move_file(
source_path,
sources_dir / source_path.name,
move=move,
)
profile_src = source_path.with_name(f"{source_path.stem}.conversion_profile.json")
if profile_src.is_file():
self._copy_or_move_file(
profile_src,
sources_dir / f"{archived_markdown.stem}.conversion_profile.json",
move=move,
)
asset_dir = self._companion_asset_dir(source_path)
if asset_dir is None:
return archived_markdown, None, None
imported_asset_dir = self._copy_or_move_tree(
asset_dir,
sources_dir / f"{archived_markdown.stem}_files",
move=move,
)
self._rewrite_markdown_asset_refs(
archived_markdown,
original_asset_dirname=asset_dir.name,
imported_asset_dirname=imported_asset_dir.name,
)
note = None
if archived_markdown.stem != source_path.stem:
note = (
f"{source_path}: renamed imported markdown to {archived_markdown.name} "
f"and rewrote asset references to {imported_asset_dir.name}/"
)
return archived_markdown, imported_asset_dir, note
def import_sources(
self,
project_path: str,
source_items: list[str],
move: bool = False,
copy: bool = False,
) -> dict[str, list[str]]:
if move and copy:
raise ValueError("--move and --copy are mutually exclusive")
project_dir = Path(project_path)
if not project_dir.exists() or not project_dir.is_dir():
raise FileNotFoundError(f"Project directory not found: {project_dir}")
if not source_items:
raise ValueError("At least one source path or URL is required")
sources_dir = self._source_dir(project_dir)
summary: dict[str, list[str]] = {
"archived": [],
"url_records": [],
"markdown": [],
"assets": [],
"images": [],
"analysis": [],
"notes": [],
"skipped": [],
}
expanded_items: list[str] = []
supplied_dirs: list[Path] = []
for item in source_items:
if is_url(item):
expanded_items.append(item)
continue
item_path = Path(item)
if item_path.is_dir():
supplied_dirs.append(item_path)
directory_files = sorted(
path for path in item_path.iterdir() if path.is_file()
)
if directory_files:
expanded_items.extend(str(path) for path in directory_files)
summary["notes"].append(
f"{item}: expanded directory into {len(directory_files)} file(s)"
)
else:
summary["skipped"].append(f"{item}: directory contains no files")
continue
expanded_items.append(item)
explicit_markdown_stems = {
Path(item).stem
for item in expanded_items
if not is_url(item)
and Path(item).exists()
and Path(item).is_file()
and Path(item).suffix.lower() in {".md", ".markdown"}
}
for item in expanded_items:
if is_url(item):
markdown_path = self._ensure_unique_path(
sources_dir / f"{derive_url_basename(item)}.md"
)
try:
self._import_url(item, markdown_path)
except Exception as exc: # pragma: no cover - summary path
archived = self._archive_url_record(sources_dir, item)
summary["url_records"].append(str(archived))
summary["skipped"].append(f"{item}: {exc}")
continue
if not self._is_valid_imported_url_markdown(markdown_path):
markdown_path.unlink(missing_ok=True)
archived = self._archive_url_record(sources_dir, item)
summary["url_records"].append(str(archived))
summary["skipped"].append(f"{item}: URL conversion produced no usable Markdown")
continue
summary["markdown"].append(str(markdown_path))
self._propagate_companion_image_assets(markdown_path, project_dir)
continue
source_path = Path(item)
if not source_path.exists():
summary["skipped"].append(f"{item}: path not found")
continue
if source_path.is_dir():
summary["skipped"].append(f"{item}: directories are not supported")
continue
inside_projects = is_within_path(source_path, PROJECTS_ROOT)
if copy:
effective_move = False
elif inside_projects:
effective_move = True
else:
effective_move = False
if move and not inside_projects:
print(
f"note: {source_path} is outside {PROJECTS_ROOT}; copied "
f"(not moved). Only sources under projects/ may be moved.",
file=sys.stderr,
)
elif inside_projects and not move and not copy:
print(
f"note: {source_path} is under projects/; moved into the target "
f"project. Pass --copy to preserve it.",
file=sys.stderr,
)
suffix = source_path.suffix.lower()
if suffix in {".md", ".markdown"}:
duplicate_markdown = self._find_equivalent_markdown(source_path, sources_dir)
if duplicate_markdown is not None:
summary["markdown"].append(str(duplicate_markdown))
self._propagate_companion_image_assets(duplicate_markdown, project_dir)
summary["notes"].append(
f"{item}: skipped duplicate markdown import because equivalent content already exists as {duplicate_markdown.name}"
)
continue
archived_markdown, asset_dir, note = self._import_markdown_with_assets(
source_path,
sources_dir,
move=effective_move,
)
summary["archived"].append(str(archived_markdown))
summary["markdown"].append(str(archived_markdown))
if asset_dir is not None:
summary["assets"].append(str(asset_dir))
self._propagate_image_assets(asset_dir, project_dir)
if note:
summary["notes"].append(note)
continue
archived_path = self._copy_or_move_file(
source_path,
sources_dir / source_path.name,
move=effective_move,
)
summary["archived"].append(str(archived_path))
if suffix in BITMAP_IMAGE_SUFFIXES:
images_dir = project_dir / "images"
images_dir.mkdir(parents=True, exist_ok=True)
image_path = self._ensure_unique_path(images_dir / archived_path.name)
shutil.copy2(archived_path, image_path)
summary["images"].append(str(image_path))
if image_path.name != archived_path.name:
summary["notes"].append(
f"{item}: copied runtime image as {image_path.name} "
"to avoid a filename collision"
)
elif suffix in PDF_SUFFIXES:
canonical_markdown_path = sources_dir / f"{archived_path.stem}.md"
if archived_path.stem in explicit_markdown_stems:
summary["notes"].append(
f"{item}: skipped PDF auto-conversion because a same-stem Markdown source was provided"
)
continue
if canonical_markdown_path.exists():
summary["markdown"].append(str(canonical_markdown_path))
self._propagate_companion_image_assets(canonical_markdown_path, project_dir)
summary["notes"].append(
f"{item}: skipped PDF auto-conversion because {canonical_markdown_path.name} already exists"
)
continue
markdown_path = canonical_markdown_path
try:
self._import_pdf(archived_path, markdown_path)
summary["markdown"].append(str(markdown_path))
self._propagate_companion_image_assets(markdown_path, project_dir)
except Exception as exc: # pragma: no cover - summary path
summary["skipped"].append(f"{item}: PDF conversion failed ({exc})")
elif suffix in PRESENTATION_SUFFIXES:
canonical_markdown_path = sources_dir / f"{archived_path.stem}.md"
try:
intake_dir = self._import_pptx_intake(archived_path, project_dir)
intake_str = str(intake_dir)
if intake_str not in summary["analysis"]:
summary["analysis"].append(intake_str)
except Exception as exc: # pragma: no cover - summary path
summary["notes"].append(f"{item}: PPTX intake analysis failed ({exc})")
if archived_path.stem in explicit_markdown_stems:
summary["notes"].append(
f"{item}: skipped presentation auto-conversion because a same-stem Markdown source was provided"
)
continue
if canonical_markdown_path.exists():
summary["markdown"].append(str(canonical_markdown_path))
self._propagate_companion_image_assets(canonical_markdown_path, project_dir)
summary["notes"].append(
f"{item}: skipped presentation auto-conversion because {canonical_markdown_path.name} already exists"
)
continue
markdown_path = canonical_markdown_path
try:
self._import_presentation(archived_path, markdown_path)
summary["markdown"].append(str(markdown_path))
self._propagate_companion_image_assets(markdown_path, project_dir)
except Exception as exc: # pragma: no cover - summary path
summary["skipped"].append(f"{item}: presentation conversion failed ({exc})")
elif suffix in EXCEL_SUFFIXES:
canonical_markdown_path = sources_dir / f"{archived_path.stem}.md"
if archived_path.stem in explicit_markdown_stems:
summary["notes"].append(
f"{item}: skipped Excel auto-conversion because a same-stem Markdown source was provided"
)
continue
if canonical_markdown_path.exists():
summary["markdown"].append(str(canonical_markdown_path))
self._propagate_companion_image_assets(canonical_markdown_path, project_dir)
summary["notes"].append(
f"{item}: skipped Excel auto-conversion because {canonical_markdown_path.name} already exists"
)
continue
markdown_path = canonical_markdown_path
try:
self._import_excel(archived_path, markdown_path)
summary["markdown"].append(str(markdown_path))
self._propagate_companion_image_assets(markdown_path, project_dir)
except Exception as exc: # pragma: no cover - summary path
summary["skipped"].append(f"{item}: Excel conversion failed ({exc})")
elif suffix in LEGACY_EXCEL_SUFFIXES:
summary["notes"].append(
f"{item}: archived only; legacy .xls is not converted automatically. "
"Resave as .xlsx to generate Markdown."
)
elif suffix in TABLE_TEXT_SUFFIXES:
summary["notes"].append(
f"{item}: archived as a plain-text table source; no Markdown conversion needed"
)
elif suffix in DOC_SUFFIXES:
canonical_markdown_path = sources_dir / f"{archived_path.stem}.md"
if archived_path.stem in explicit_markdown_stems:
summary["notes"].append(
f"{item}: skipped document auto-conversion because a same-stem Markdown source was provided"
)
continue
if canonical_markdown_path.exists():
summary["markdown"].append(str(canonical_markdown_path))
self._propagate_companion_image_assets(canonical_markdown_path, project_dir)
summary["notes"].append(
f"{item}: skipped document auto-conversion because {canonical_markdown_path.name} already exists"
)
continue
markdown_path = canonical_markdown_path
try:
self._import_doc(archived_path, markdown_path)
summary["markdown"].append(str(markdown_path))
self._propagate_companion_image_assets(markdown_path, project_dir)
except Exception as exc: # pragma: no cover - summary path
summary["skipped"].append(f"{item}: document conversion failed ({exc})")
elif suffix == ".txt":
markdown_path = self._normalize_text_source(archived_path, sources_dir)
summary["markdown"].append(str(markdown_path))
else:
summary["notes"].append(f"{item}: archived only, no automatic conversion")
# Cleanup: only a projects-local source directory may be removed after
# its files move into the target project. Every other location is copied
# and remains untouched, even when the caller passes --move.
for directory in supplied_dirs:
if copy or not is_within_path(directory, PROJECTS_ROOT):
continue
if directory.is_dir() and not any(directory.iterdir()):
try:
directory.rmdir()
except OSError:
continue
summary["notes"].append(
f"{directory}: removed empty source directory after import"
)
return summary
def validate_project(self, project_path: str) -> tuple[bool, list[str], list[str]]:
project_path_obj = Path(project_path)
_, errors, warnings = validate_project_structure(
str(project_path_obj),
validate_communication=False,
)
if project_path_obj.exists() and project_path_obj.is_dir():
project_info = get_project_info_common(str(project_path_obj))
artifact_errors, artifact_warnings = validate_project_artifacts(
project_path_obj,
project_info,
)
errors.extend(artifact_errors)
warnings.extend(artifact_warnings)
if project_path_obj.exists() and project_path_obj.is_dir():
info = get_project_info_common(str(project_path_obj))
if info.get("svg_files"):
svg_files = [project_path_obj / "svg_output" / name for name in info["svg_files"]]
expected_format = info.get("format")
if expected_format == "unknown":
expected_format = None
warnings.extend(validate_svg_viewbox(svg_files, expected_format))
return not errors, list(dict.fromkeys(errors)), warnings
def get_project_info(self, project_path: str) -> dict[str, object]:
shared = get_project_info_common(project_path)
return {
"name": shared.get("name", Path(project_path).name),
"path": shared.get("path", str(project_path)),
"exists": shared.get("exists", False),
"svg_count": shared.get("svg_count", 0),
"has_spec": shared.get("has_spec", False),
"has_source": shared.get("has_source", False),
"source_count": shared.get("source_count", 0),
"canvas_format": shared.get("format_name", "Unknown"),
"create_date": shared.get("date_formatted", "Unknown"),
}
def build_parser() -> argparse.ArgumentParser:
"""Build the command-line parser."""
parser = argparse.ArgumentParser(
description="PPT Master project management helpers.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""Examples:
python3 scripts/project_manager.py init demo --format ppt169
python3 scripts/project_manager.py import-sources projects/demo file.md
python3 scripts/project_manager.py scaffold-spec projects/demo_ppt169_20260718
python3 scripts/project_manager.py scaffold-lock projects/demo_ppt169_20260718
python3 scripts/project_manager.py validate projects/demo
python3 scripts/project_manager.py info projects/demo
python3 scripts/project_manager.py page-context projects/demo P07 --record-usage
python3 scripts/project_manager.py page-context-report projects/demo
""",
)
subparsers = parser.add_subparsers(dest="command", required=True)
init = subparsers.add_parser("init", help="Create a project directory")
init.add_argument("project_name", help="Project name")
init.add_argument("--format", default="ppt169", help="Canvas format (default: ppt169)")
init.add_argument("--dir", default=None, help="Base directory for the project")
init.add_argument(
"--quick-generate",
action="store_true",
help=(
"Create svg_output plus the validation workflow audit log and "
"omit README.md"
),
)
import_sources = subparsers.add_parser(
"import-sources",
help="Import source files or URLs into a project",
)
import_sources.add_argument("project_path", help="Project directory")
import_sources.add_argument("sources", nargs="+", help="Source files, directories, or URLs")
mode = import_sources.add_mutually_exclusive_group()
mode.add_argument(
"--move",
action="store_true",
help="Move local sources under projects/; sources elsewhere are copied",
)
mode.add_argument("--copy", action="store_true", help="Copy local source files")
scaffold_spec = subparsers.add_parser(
"scaffold-spec",
help="Create design_spec.md from the versioned scaffold",
)
scaffold_spec.add_argument("project_path", help="Project directory")
scaffold_lock = subparsers.add_parser(
"scaffold-lock",
help="Create spec_lock.md from the versioned scaffold",
)
scaffold_lock.add_argument("project_path", help="Project directory")
validate = subparsers.add_parser("validate", help="Validate a project directory")
validate.add_argument("project_path", help="Project directory")
info = subparsers.add_parser("info", help="Print project metadata")
info.add_argument("project_path", help="Project directory")
page_context = subparsers.add_parser(
"page-context",
help="Print one deterministic per-page execution view",
)
page_context.add_argument("project_path", help="Project directory")
page_context.add_argument("page", help="Positive page key such as P07")
page_context.add_argument(
"--bundle",
action="store_true",
help="Deprecated compatibility flag; output remains compact",
)
page_context.add_argument(
"--pretty",
action="store_true",
help="Pretty-print the page-context JSON payload",
)
page_context.add_argument(
"--record-usage",
action="store_true",
help="Write compact-output token telemetry under analysis/page-context/",
)
page_context_report = subparsers.add_parser(
"page-context-report",
help="Summarize fresh per-page context telemetry",
)
page_context_report.add_argument("project_path", help="Project directory")
return parser
def main(argv: list[str] | None = None) -> int:
"""Run the CLI entry point."""
require_skill_integrity()
parser = build_parser()
args = parser.parse_args(argv)
manager = ProjectManager()
try:
if args.command == "init":
project_path = manager.init_project(
args.project_name,
args.format,
base_dir=args.dir,
quick_generate=args.quick_generate,
)
print(f"[OK] Project initialized: {project_path}")
print("Next:")
if args.quick_generate:
print("1. Generate SVG files into svg_output/")
print("2. Run the Quick Generate final checker and exporter")
profile = "quick"
else:
print("1. Put source files into sources/ (or use import-sources)")
print("2. Save your design spec to the project root")
print("3. Generate SVG files into svg_output/")
profile = "default"
try:
append_note(
project_path,
f"Project initialized: profile={profile}; "
f"canvas={args.format}; path={project_path}",
)
except OSError as exc:
print(
f"[WARN] Workflow audit unavailable: {exc}",
file=sys.stderr,
)
return 0
if args.command == "import-sources":
summary = manager.import_sources(
args.project_path,
args.sources,
move=args.move,
copy=args.copy,
)
has_usable_import = _has_usable_import(summary)
if has_usable_import:
print(f"[OK] Imported sources into: {args.project_path}")
else:
print(
f"[ERROR] No usable sources imported into: {args.project_path}",
file=sys.stderr,
)
if summary["archived"]:
print("\nArchived originals:")
for item in summary["archived"]:
print(f" - {item}")
if summary["url_records"]:
print("\nArchived URL records:")
for item in summary["url_records"]:
print(f" - {item}")
if summary["markdown"]:
print("\nNormalized markdown:")
for item in summary["markdown"]:
print(f" - {item}")
if summary["assets"]:
print("\nImported asset directories:")
for item in summary["assets"]:
print(f" - {item}")
if summary["images"]:
print("\nRuntime image copies:")
for item in summary["images"]:
print(f" - {item}")
if summary["analysis"]:
print("\nAnalysis artifacts:")
for item in summary["analysis"]:
print(f" - {item}")
if summary["notes"]:
print("\nNotes:")
for item in summary["notes"]:
print(f" - {item}")
if summary["skipped"]:
print("\nSkipped:")
for item in summary["skipped"]:
print(f" - {item}")
return 0 if has_usable_import else 1
if args.command == "scaffold-spec":
artifact_path = manager.scaffold_artifact(args.project_path, "design_spec")
print(f"[OK] Design spec scaffold created: {artifact_path}")
return 0
if args.command == "scaffold-lock":
artifact_path = manager.scaffold_artifact(args.project_path, "spec_lock")
print(f"[OK] Execution lock scaffold created: {artifact_path}")
return 0
if args.command == "validate":
project_path = args.project_path
is_valid, errors, warnings = manager.validate_project(project_path)
print(f"\nProject validation: {project_path}")
print("=" * 60)
if errors:
print("\n[ERROR]")
for error in errors:
print(f" - {error}")
if warnings:
print("\n[WARN]")
for warning in warnings:
print(f" - {warning}")
if is_valid and not warnings:
print("\n[OK] Project structure is complete.")
elif is_valid:
print("\n[OK] Project structure is valid, with warnings.")
else:
print("\n[ERROR] Project structure is invalid.")
return 1
return 0
if args.command == "info":
project_path = args.project_path
info = manager.get_project_info(project_path)
print(f"\nProject info: {info['name']}")
print("=" * 60)
print(f"Path: {info['path']}")
print(f"Exists: {'Yes' if info['exists'] else 'No'}")
print(f"SVG files: {info['svg_count']}")
print(f"Design spec: {'Yes' if info['has_spec'] else 'No'}")
print(f"Source materials: {'Yes' if info['has_source'] else 'No'}")
print(f"Source count: {info['source_count']}")
print(f"Canvas format: {info['canvas_format']}")
print(f"Created: {info['create_date']}")
return 0
if args.command == "page-context":
result = build_page_context(args.project_path, args.page)
output, measured_reads = render_page_context(
result,
bundle=args.bundle,
pretty=args.pretty,
)
if args.record_usage:
_usage_path, token_status = record_page_context_usage(
result,
output,
measured_reads,
)
if token_status != "exact":
print(
"[WARN] tiktoken/o200k_base unavailable; recorded bytes "
"and hashes without token counts",
file=sys.stderr,
)
print(output, end="")
return 0
if args.command == "page-context-report":
report = page_context_usage_report(args.project_path)
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0
parser.error(f"Unknown command: {args.command}")
except Exception as exc:
print(f"[ERROR] {exc}")
return 1