#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File: src/scitex_writer/writer.py
"""
Writer class for manuscript LaTeX compilation.
Provides object-oriented interface to scitex-writer functionality.
"""
from __future__ import annotations
import logging
import shutil
import subprocess
from pathlib import Path
from typing import Callable, Optional
from ._compile import (
CompilationResult,
compile_manuscript,
compile_revision,
compile_supplementary,
)
from ._dataclasses import ManuscriptTree, RevisionTree, SupplementaryTree
from ._dataclasses.config import DOC_TYPE_DIRS
from ._dataclasses.tree import ScriptsTree, SharedTree
from ._project._create import clone_writer_project
from ._utils._watch import watch_manuscript
logger = logging.getLogger(__name__)
def _find_git_root(project_dir: Path) -> Optional[Path]:
"""Find git root for project directory."""
try:
result = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
cwd=str(project_dir),
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0:
return Path(result.stdout.strip())
except Exception:
pass
return None
class Writer:
"""LaTeX manuscript compiler."""
def __init__(
self,
project_dir: Path,
name: Optional[str] = None,
git_strategy: Optional[str] = "child",
branch: Optional[str] = None,
tag: Optional[str] = None,
):
"""
Initialize for project directory.
If directory doesn't exist, creates new project.
Parameters
----------
project_dir : Path
Path to project directory.
name : str, optional
Project name (used if creating new project).
git_strategy : str, optional
Git initialization strategy:
- 'child': Create isolated git in project directory (default)
- 'parent': Use parent git repository
- 'origin': Preserve template's original git history
- None or 'none': Disable git initialization
branch : str, optional
Specific branch of template repository to clone.
If None, clones the default branch. Mutually exclusive with tag.
tag : str, optional
Specific tag/release of template repository to clone.
If None, clones the default branch. Mutually exclusive with branch.
"""
self.project_name = name or Path(project_dir).name
self.project_dir = Path(project_dir)
self.git_strategy = git_strategy
self.branch = branch
self.tag = tag
ref_info = ""
if branch:
ref_info = f" (branch: {branch})"
elif tag:
ref_info = f" (tag: {tag})"
logger.info(
f"Writer: Initializing with:\n"
f" Project Name: {self.project_name}\n"
f" Project Directory: {self.project_dir.absolute()}\n"
f" Git Strategy: {self.git_strategy}{ref_info}..."
)
# Create or attach to project
self.project_dir = self._attach_or_create_project(name)
# Find git root (may be the project dir or a parent)
self.git_root = _find_git_root(self.project_dir) or self.project_dir
# Document accessors (pass git_root for efficiency)
self.shared = SharedTree(self.project_dir / "00_shared", git_root=self.git_root)
self.manuscript = ManuscriptTree(
self.project_dir / "01_manuscript", git_root=self.git_root
)
self.supplementary = SupplementaryTree(
self.project_dir / "02_supplementary", git_root=self.git_root
)
self.revision = RevisionTree(
self.project_dir / "03_revision", git_root=self.git_root
)
self.scripts = ScriptsTree(self.project_dir / "scripts", git_root=self.git_root)
logger.info(f"Writer: Initialization complete for {self.project_name}")
def _attach_or_create_project(self, name: Optional[str] = None) -> Path:
"""
Create new project or attach to existing one.
If project directory doesn't exist, creates it based on git_strategy:
- 'child': Full template with git initialization
- 'parent'/'None': Minimal directory structure
Parameters
----------
name : str, optional
Project name (used if creating new project).
Returns
-------
Path
Path to the project directory.
"""
if self.project_dir.exists():
logger.info(
f"Writer: Attached to existing project at {self.project_dir.absolute()}"
)
# Verify existing project structure
self._verify_project_structure()
return self.project_dir
project_name = name or self.project_dir.name
logger.info(
f"Writer: Creating new project '{project_name}' at {self.project_dir.absolute()}"
)
# Initialize project directory structure
success = clone_writer_project(
str(self.project_dir), self.git_strategy, self.branch, self.tag
)
if not success:
logger.error(
f"Writer: Failed to initialize project directory for {project_name}"
)
raise RuntimeError(
f"Could not create project directory at {self.project_dir}"
)
# Verify project directory was created
if not self.project_dir.exists():
logger.error(
f"Writer: Project directory {self.project_dir} was not created"
)
raise RuntimeError(f"Project directory {self.project_dir} was not created")
logger.info(
f"Writer: Successfully created project at {self.project_dir.absolute()}"
)
return self.project_dir
def _verify_project_structure(self) -> None:
"""
Verify attached project has expected structure.
Checks:
- Required directories exist (01_manuscript, 02_supplementary, 03_revision)
Raises
------
RuntimeError
If structure is invalid.
"""
required_dirs = [
self.project_dir / "01_manuscript",
self.project_dir / "02_supplementary",
self.project_dir / "03_revision",
]
for dir_path in required_dirs:
if not dir_path.exists():
logger.error(f"Writer: Expected directory missing: {dir_path}")
raise RuntimeError(
f"Project structure invalid: missing {dir_path.name} directory"
)
logger.info(
f"Writer: Project structure verified at {self.project_dir.absolute()}"
)
def compile_manuscript(
self,
timeout: int = 300,
log_callback: Optional[Callable[[str], None]] = None,
progress_callback: Optional[Callable[[int, str], None]] = None,
) -> CompilationResult:
"""
Compile manuscript to PDF with optional live callbacks.
Runs scripts/shell/compile_manuscript.sh with configured settings.
Parameters
----------
timeout : int, optional
Maximum compilation time in seconds (default: 300).
log_callback : callable, optional
Called with each log line: log_callback("Running pdflatex...").
progress_callback : callable, optional
Called with progress: progress_callback(50, "Pass 2/3").
Returns
-------
CompilationResult
With success status, PDF path, and errors/warnings.
Examples
--------
>>> writer = Writer(Path("my_paper"))
>>> result = writer.compile_manuscript()
>>> if result.success:
... print(f"PDF created: {result.output_pdf}")
"""
return compile_manuscript(
self.project_dir,
timeout=timeout,
log_callback=log_callback,
progress_callback=progress_callback,
)
def compile_supplementary(
self,
timeout: int = 300,
log_callback: Optional[Callable[[str], None]] = None,
progress_callback: Optional[Callable[[int, str], None]] = None,
) -> CompilationResult:
"""
Compile supplementary materials to PDF with optional live callbacks.
Runs scripts/shell/compile_supplementary.sh with configured settings.
Parameters
----------
timeout : int, optional
Maximum compilation time in seconds (default: 300).
log_callback : callable, optional
Called with each log line.
progress_callback : callable, optional
Called with progress updates.
Returns
-------
CompilationResult
With success status, PDF path, and errors/warnings.
Examples
--------
>>> writer = Writer(Path("my_paper"))
>>> result = writer.compile_supplementary()
>>> if result.success:
... print(f"PDF created: {result.output_pdf}")
"""
return compile_supplementary(
self.project_dir,
timeout=timeout,
log_callback=log_callback,
progress_callback=progress_callback,
)
def compile_revision(
self,
track_changes: bool = False,
timeout: int = 300,
log_callback: Optional[Callable[[str], None]] = None,
progress_callback: Optional[Callable[[int, str], None]] = None,
) -> CompilationResult:
"""
Compile revision document with optional change tracking and live callbacks.
Runs scripts/shell/compile_revision.sh with configured settings.
Parameters
----------
track_changes : bool, optional
Enable change tracking in compiled PDF (default: False).
timeout : int, optional
Maximum compilation time in seconds (default: 300).
log_callback : callable, optional
Called with each log line.
progress_callback : callable, optional
Called with progress updates.
Returns
-------
CompilationResult
With success status, PDF path, and errors/warnings.
Examples
--------
>>> writer = Writer(Path("my_paper"))
>>> result = writer.compile_revision(track_changes=True)
>>> if result.success:
... print(f"Revision PDF: {result.output_pdf}")
"""
return compile_revision(
self.project_dir,
track_changes=track_changes,
timeout=timeout,
log_callback=log_callback,
progress_callback=progress_callback,
)
def get_section(self, section_name: str, doc_type: str = "manuscript"):
"""Get a DocumentSection by name and document type.
Parameters
----------
section_name : str
Section name (e.g., 'abstract', 'introduction', 'title').
doc_type : str
Document type: 'shared', 'manuscript', 'supplementary', or 'revision'.
Returns
-------
DocumentSection
Section object with .read(), .write(), .commit(), .history(), .diff().
Raises
------
ValueError
If doc_type is unknown or section_name not found.
"""
doc_map = {
"shared": self.shared,
"manuscript": self.manuscript,
"supplementary": self.supplementary,
"revision": self.revision,
}
if doc_type not in doc_map:
raise ValueError(
f"Unknown doc_type: {doc_type}. Valid: {list(doc_map.keys())}"
)
doc = doc_map[doc_type]
if doc_type == "shared":
if not hasattr(doc, section_name):
raise ValueError(
f"Section '{section_name}' not in shared. "
f"Available: {self._list_sections(doc)}"
)
return getattr(doc, section_name)
else:
if not hasattr(doc.contents, section_name):
raise ValueError(
f"Section '{section_name}' not in {doc_type}. "
f"Available: {self._list_sections(doc.contents)}"
)
return getattr(doc.contents, section_name)
def read_section(self, section_name: str, doc_type: str = "manuscript") -> str:
"""Read a section's content as string.
Parameters
----------
section_name : str
Section name (e.g., 'abstract', 'introduction').
doc_type : str
Document type: 'shared', 'manuscript', 'supplementary', or 'revision'.
Returns
-------
str
Section content. Empty string if section file is empty or missing.
Examples
--------
>>> writer = Writer(Path("my_paper"))
>>> text = writer.read_section("abstract")
>>> text = writer.read_section("title", "shared")
"""
section = self.get_section(section_name, doc_type)
content = section.read()
if isinstance(content, list):
content = "\n".join(content)
return content or ""
def write_section(
self, section_name: str, content: str, doc_type: str = "manuscript"
) -> bool:
"""Write content to a section.
Parameters
----------
section_name : str
Section name (e.g., 'abstract', 'introduction').
content : str
Content to write.
doc_type : str
Document type: 'shared', 'manuscript', 'supplementary', or 'revision'.
Returns
-------
bool
True if write succeeded.
Examples
--------
>>> writer = Writer(Path("my_paper"))
>>> writer.write_section("abstract", "New abstract text.")
True
"""
section = self.get_section(section_name, doc_type)
return section.write(content)
def _list_sections(self, tree_or_contents) -> list:
"""List available section names from a tree or contents object."""
return [
attr
for attr in dir(tree_or_contents)
if not attr.startswith("_")
and hasattr(getattr(tree_or_contents, attr, None), "path")
]
def watch(self, on_compile: Optional[Callable] = None) -> None:
"""Auto-recompile on file changes."""
watch_manuscript(self.project_dir, on_compile=on_compile)
def get_pdf(self, doc_type: str = "manuscript") -> Optional[Path]:
"""Get output PDF path (Read)."""
pdf = self.project_dir / DOC_TYPE_DIRS[doc_type] / f"{doc_type}.pdf"
return pdf if pdf.exists() else None
def delete(self) -> bool:
"""Delete project directory (Delete)."""
try:
logger.warning(
f"Writer: Deleting project directory at {self.project_dir.absolute()}"
)
shutil.rmtree(self.project_dir)
logger.info(
f"Writer: Successfully deleted project at {self.project_dir.absolute()}"
)
return True
except Exception as e:
logger.error(
f"Writer: Failed to delete project directory at {self.project_dir.absolute()}: {e}"
)
return False
__all__ = ["Writer"]
# EOF
/