potaserandClaude Opus 5
e641c8b9ab
【轻量拓展分析】 - 新增 gfa_core/extension_stage_lightweight/ 核心模块，支持用户上传代码/数据/原始命令 - 前端卡片式交互流程：候选核对 → 验证 → 结果审查 → 授权发布 → 运行 - Celery 不受限任务 + 轮询机制 + sessionStorage 操作恢复 - 启动脚本新增 BBAV5_LIGHTWEIGHT_* 环境变量 【WGS2GWAS 支持】 - 新增 13 个 WGS 阶段定义（QC → MAF/Missing Filter） - email_haplotype_region.py 新增 LTR-RT 分析功能 【结果区域重构】 - 单层折叠 → 三组双层折叠：RNA-seq / 基因家族 / WGS2GWAS - 结果按执行时间排序，保留最新尝试 【物种角色解析增强】 - 新增 _resolve_subfamily_request_species_roles() - 支持 reference 角色的识别 - 直接 stage 请求保留 intent 中的物种角色 【会话 Job 隔离】 - rebind_job() 跨 job 时创建新会话，返回新 conversation_id 【前端交互增强】 - RAG 文献问答模式 - NL 触发 config/samples 自动填写、亚家族确认表、基因家族成员确认表 - 执行状态从后端 resource_tasks 恢复 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
568 lines
21 KiB
Python
Executable File
568 lines
21 KiB
Python
Executable File
"""Gene-family analysis scope resolution.
|
|
|
|
The legacy pipeline uses ``prefix`` as both an output namespace and an
|
|
analysis-range identifier. This module keeps ``prefix`` for compatibility and
|
|
adds an explicit scope object that records family, species roles, selected
|
|
species and resolved input files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
from typing import Any, Iterable
|
|
|
|
|
|
ROLE_TARGET = "target"
|
|
ROLE_REFERENCE = "reference"
|
|
ROLE_COMPARATIVE = "comparative"
|
|
ROLE_SEED_REFERENCE = "seed_reference"
|
|
|
|
|
|
@dataclass
|
|
class ScopeFile:
|
|
original_file_name: str
|
|
species: str
|
|
file_type: str
|
|
role: str
|
|
file_scope: str = ""
|
|
gene_family: str = ""
|
|
ref: bool = False
|
|
reference_use: str = ""
|
|
path: str = ""
|
|
exists: bool = False
|
|
size: int | None = None
|
|
mtime: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class GeneFamilyScope:
|
|
family: str
|
|
prefix: str
|
|
scope_id: str
|
|
scope_mode: str
|
|
target_species: list[str] = field(default_factory=list)
|
|
reference_species: list[str] = field(default_factory=list)
|
|
primary_reference_species: list[str] = field(default_factory=list)
|
|
identify_blast_reference_species: list[str] = field(default_factory=list)
|
|
comparative_species: list[str] = field(default_factory=list)
|
|
selected_species: list[str] = field(default_factory=list)
|
|
config_path: str = ""
|
|
files: dict[str, dict[str, list[dict[str, Any]]]] = field(default_factory=dict)
|
|
analysis: dict[str, str] = field(default_factory=dict)
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def clean_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
text = str(value).strip()
|
|
if text.lower() in {"none", "nan", "null"}:
|
|
return ""
|
|
return text
|
|
|
|
|
|
def split_csv(value: Any) -> list[str]:
|
|
text = clean_text(value)
|
|
if not text:
|
|
return []
|
|
items = re.split(r"[,;，、\s]+", text)
|
|
return unique_keep_order(item.strip() for item in items if item.strip())
|
|
|
|
|
|
def unique_keep_order(items: Iterable[str]) -> list[str]:
|
|
seen: set[str] = set()
|
|
ordered: list[str] = []
|
|
for item in items:
|
|
if item and item not in seen:
|
|
seen.add(item)
|
|
ordered.append(item)
|
|
return ordered
|
|
|
|
|
|
def parse_bool(value: Any) -> bool:
|
|
text = clean_text(value).lower()
|
|
return text in {"1", "true", "ture", "yes", "y", "ref"}
|
|
|
|
|
|
def sanitize_token(value: str) -> str:
|
|
token = re.sub(r"[^A-Za-z0-9_.-]+", "-", clean_text(value))
|
|
token = token.strip(".-_")
|
|
return token or "NA"
|
|
|
|
|
|
|
|
def normalize_reference_use(value: Any) -> str:
|
|
text = clean_text(value).lower().replace("-", "_").replace(" ", "_")
|
|
aliases = {
|
|
"blast": "identify_blast",
|
|
"identify": "identify_blast",
|
|
"identify_reference": "identify_blast",
|
|
"identify_blast_reference": "identify_blast",
|
|
"homology_search": "identify_blast",
|
|
"primary": "downstream_primary",
|
|
"main_reference": "downstream_primary",
|
|
"primary_reference": "downstream_primary",
|
|
"downstream": "downstream_primary",
|
|
"downstream_reference": "downstream_primary",
|
|
"all": "both",
|
|
"both_reference": "both",
|
|
"no": "none",
|
|
"false": "none",
|
|
}
|
|
return aliases.get(text, text)
|
|
|
|
def reference_use_includes(value: Any, purpose: str) -> bool:
|
|
normalized = normalize_reference_use(value)
|
|
if normalized == "both":
|
|
return True
|
|
if purpose == "identify_blast":
|
|
return normalized in {"identify_blast"}
|
|
if purpose == "downstream_primary":
|
|
return normalized in {"downstream_primary"}
|
|
return False
|
|
|
|
def normalize_role(role: Any, ref: Any = None, file_scope: Any = None) -> str:
|
|
text = clean_text(role).lower().replace("-", "_")
|
|
scope = clean_text(file_scope).lower().replace("-", "_")
|
|
if text in {"target", "targets"}:
|
|
return ROLE_TARGET
|
|
if text in {"comparative", "comparison", "compare", "outgroup"}:
|
|
return ROLE_COMPARATIVE
|
|
if text in {"seed_reference", "seed", "known_reference"}:
|
|
return ROLE_SEED_REFERENCE
|
|
if text in {"ref", "reference", "references"}:
|
|
if scope == "known_family":
|
|
return ROLE_SEED_REFERENCE
|
|
return ROLE_REFERENCE
|
|
if parse_bool(ref):
|
|
if scope == "known_family":
|
|
return ROLE_SEED_REFERENCE
|
|
return ROLE_REFERENCE
|
|
return ""
|
|
|
|
|
|
def normalize_file_type(file_type: Any, file_scope: Any = None) -> str:
|
|
text = clean_text(file_type).lower().replace("-", "_")
|
|
scope = clean_text(file_scope).lower().replace("-", "_")
|
|
aliases = {
|
|
"protein": "pep",
|
|
"all_pep": "pep",
|
|
"proteome": "pep",
|
|
"faa": "pep",
|
|
"fa_pep": "pep",
|
|
"fasta_pep": "pep",
|
|
"annotation": "gff",
|
|
"gff3": "gff",
|
|
"gtf": "gff",
|
|
"dna": "genome",
|
|
"genomic": "genome",
|
|
"cdna": "cds",
|
|
}
|
|
text = aliases.get(text, text)
|
|
if text == "pep" and scope == "known_family":
|
|
return "family_pep"
|
|
if text == "cds" and scope == "known_family":
|
|
return "family_cds"
|
|
return text
|
|
|
|
|
|
def find_config_file(input_dir: Path) -> Path | None:
|
|
candidates = [input_dir / "config.xlsx", input_dir / "config.xlsm", input_dir / "config.xls"]
|
|
for candidate in candidates:
|
|
if candidate.exists():
|
|
return candidate
|
|
matches = sorted(input_dir.glob("*config*.xlsx"))
|
|
return matches[0] if matches else None
|
|
|
|
|
|
def read_config_workbook(config_path: Path) -> tuple[list[dict[str, Any]], dict[str, str]]:
|
|
try:
|
|
from openpyxl import load_workbook
|
|
except Exception as exc: # pragma: no cover - dependency issue is environment-specific.
|
|
raise RuntimeError("openpyxl is required to read config.xlsx") from exc
|
|
|
|
wb = load_workbook(config_path, data_only=True)
|
|
rows: list[dict[str, Any]] = []
|
|
analysis: dict[str, str] = {}
|
|
|
|
if "Analysis" in wb.sheetnames:
|
|
ws = wb["Analysis"]
|
|
headers = [clean_text(cell.value) for cell in next(ws.iter_rows(min_row=1, max_row=1))]
|
|
key_idx = headers.index("Key") if "Key" in headers else 0
|
|
value_idx = headers.index("Value") if "Value" in headers else 1
|
|
for row in ws.iter_rows(min_row=2, values_only=True):
|
|
if len(row) <= key_idx:
|
|
continue
|
|
key = clean_text(row[key_idx])
|
|
if not key:
|
|
continue
|
|
value = clean_text(row[value_idx] if len(row) > value_idx else "")
|
|
analysis[key] = value
|
|
|
|
sheet_name = "config" if "config" in wb.sheetnames else wb.sheetnames[0]
|
|
ws = wb[sheet_name]
|
|
header_values = [clean_text(cell.value) for cell in next(ws.iter_rows(min_row=1, max_row=1))]
|
|
for raw in ws.iter_rows(min_row=2, values_only=True):
|
|
row = {header_values[idx]: raw[idx] if idx < len(raw) else None for idx in range(len(header_values))}
|
|
original_name = Path(clean_text(row.get("Original_File_Name") or "")).name
|
|
if original_name.startswith("Ref_combined."):
|
|
continue
|
|
if any(clean_text(value) for value in row.values()):
|
|
rows.append(row)
|
|
|
|
return rows, analysis
|
|
|
|
|
|
def resolve_existing_path(input_dir: Path, workspace: Path | None, species: str, family: str, row: dict[str, Any], file_type: str) -> Path | None:
|
|
original = clean_text(row.get("Original_File_Name") or row.get("File_Name") or row.get("filename"))
|
|
names: list[str] = []
|
|
if original:
|
|
names.extend([
|
|
original,
|
|
Path(original).name,
|
|
str(Path(species) / Path(original).name),
|
|
])
|
|
|
|
if species:
|
|
if file_type == "pep":
|
|
names.extend([f"{species}.pep.fasta", f"{species}.pep.fa", f"{species}.protein.fasta"])
|
|
elif file_type == "family_pep":
|
|
names.extend([
|
|
f"{species}.{family}.pep.fa",
|
|
f"{species}.{family}.pep.fasta",
|
|
str(Path(species) / f"{species}.{family}.pep.fasta"),
|
|
str(Path(species) / f"{species}.{family}.pep.fa"),
|
|
])
|
|
elif file_type == "cds":
|
|
names.extend([f"{species}.cds.fasta", f"{species}.cds.fa"])
|
|
elif file_type == "family_cds":
|
|
names.extend([f"{species}.{family}.cds.fa", f"{species}.{family}.cds.fasta"])
|
|
elif file_type == "gff":
|
|
names.extend([f"{species}_final.gff3", f"{species}.gff3", f"{species}.gff"])
|
|
elif file_type == "genome":
|
|
names.extend([f"{species}.genome.fasta", f"{species}.genome.fa", f"{species}.fasta"])
|
|
|
|
search_dirs = [input_dir]
|
|
if workspace:
|
|
search_dirs.extend([workspace / "final", workspace])
|
|
for base in search_dirs:
|
|
for name in unique_keep_order(names):
|
|
candidate = base / name
|
|
if candidate.exists():
|
|
return candidate.resolve()
|
|
|
|
for base in reference_search_roots(input_dir, workspace):
|
|
for name in unique_keep_order(names):
|
|
filename = Path(name).name
|
|
if not filename:
|
|
continue
|
|
candidate = base / filename
|
|
if candidate.exists():
|
|
return candidate.resolve()
|
|
try:
|
|
matches = sorted(path for path in base.rglob(filename) if path.is_file())
|
|
except Exception:
|
|
matches = []
|
|
if matches:
|
|
return matches[0].resolve()
|
|
return None
|
|
|
|
|
|
def reference_search_roots(input_dir: Path, workspace: Path | None = None) -> list[Path]:
|
|
"""Local reference-data roots used to fill config paths without user prompts."""
|
|
candidates: list[Path] = []
|
|
for base in [input_dir, *list(input_dir.parents)[:4]]:
|
|
candidates.extend([
|
|
base / "ref_data",
|
|
base / "reference",
|
|
base / "references",
|
|
base / "data" / "ref_data",
|
|
])
|
|
if workspace:
|
|
for base in [workspace, *list(workspace.parents)[:4]]:
|
|
candidates.extend([
|
|
base / "ref_data",
|
|
base / "reference",
|
|
base / "references",
|
|
base / "data" / "ref_data",
|
|
])
|
|
configured_ref = os.environ.get("PLANTAI_REF_DATA_ROOT")
|
|
if configured_ref:
|
|
candidates.append(Path(configured_ref).expanduser())
|
|
|
|
roots: list[Path] = []
|
|
seen: set[str] = set()
|
|
for candidate in candidates:
|
|
try:
|
|
resolved = candidate.resolve()
|
|
except Exception:
|
|
continue
|
|
key = str(resolved)
|
|
if key in seen or not resolved.exists() or not resolved.is_dir():
|
|
continue
|
|
seen.add(key)
|
|
roots.append(resolved)
|
|
return roots
|
|
|
|
|
|
def file_record(path: Path | None) -> tuple[str, bool, int | None, float | None]:
|
|
if not path or not path.exists():
|
|
return "", False, None, None
|
|
stat = path.stat()
|
|
return str(path), True, stat.st_size, stat.st_mtime
|
|
|
|
|
|
def infer_family(prefix: str, analysis: dict[str, str], rows: list[dict[str, Any]], family_override: str = "") -> str:
|
|
if clean_text(family_override):
|
|
return clean_text(family_override)
|
|
for key in ("Family", "Gene_Family", "family", "gene_family"):
|
|
if clean_text(analysis.get(key)):
|
|
return clean_text(analysis.get(key))
|
|
if "." in clean_text(prefix):
|
|
return clean_text(prefix).split(".", 1)[1]
|
|
for row in rows:
|
|
family = clean_text(row.get("Gene_Family"))
|
|
if family:
|
|
return family
|
|
return ""
|
|
|
|
|
|
def default_prefix(prefix: str, family: str, target_species: list[str]) -> str:
|
|
provided = clean_text(prefix)
|
|
if provided:
|
|
return provided
|
|
if family and len(target_species) == 1:
|
|
return f"{target_species[0]}.{family}"
|
|
if family:
|
|
return f"Target.{family}"
|
|
return "Target"
|
|
|
|
|
|
def build_scope_id(family: str, targets: list[str], refs: list[str], comps: list[str], mode: str) -> str:
|
|
parts = [sanitize_token(family or "family")]
|
|
if targets:
|
|
parts.append("target-" + "-".join(sanitize_token(x) for x in targets))
|
|
if refs:
|
|
parts.append("ref-" + "-".join(sanitize_token(x) for x in refs))
|
|
if comps:
|
|
parts.append("comp-" + "-".join(sanitize_token(x) for x in comps))
|
|
if mode and mode != "custom":
|
|
parts.append(sanitize_token(mode))
|
|
return ".".join(parts)
|
|
|
|
|
|
def scope_hash(scope_id: str, length: int = 8) -> str:
|
|
text = sanitize_token(scope_id)
|
|
return hashlib.sha1(text.encode("utf-8")).hexdigest()[: max(4, int(length or 8))]
|
|
|
|
|
|
def summarize_species_list(role: str, species: list[str], max_items: int = 3) -> str:
|
|
clean = [sanitize_token(x) for x in species if clean_text(x)]
|
|
if not clean:
|
|
return ""
|
|
head = clean[:max_items]
|
|
suffix = f"-plus{len(clean) - max_items}" if len(clean) > max_items else ""
|
|
return f"{sanitize_token(role)}-" + "-".join(head) + suffix
|
|
|
|
|
|
def build_scope_summary(
|
|
family: str,
|
|
targets: list[str],
|
|
refs: list[str],
|
|
comps: list[str],
|
|
mode: str,
|
|
max_items: int = 3,
|
|
) -> str:
|
|
"""Human-readable, bounded scope label for filenames/UI.
|
|
|
|
The full machine scope remains in ``scope_id`` and scope JSON. This
|
|
summary intentionally avoids listing every species when users analyze many
|
|
species.
|
|
"""
|
|
parts: list[str] = []
|
|
for role, values in (("target", targets), ("ref", refs), ("comp", comps)):
|
|
part = summarize_species_list(role, values, max_items=max_items)
|
|
if part:
|
|
parts.append(part)
|
|
if not parts:
|
|
parts.append(sanitize_token(family or "family"))
|
|
if mode and mode != "custom":
|
|
parts.append(sanitize_token(mode))
|
|
return ".".join(parts)
|
|
|
|
|
|
def selected_species_for_mode(targets: list[str], refs: list[str], comps: list[str], mode: str) -> list[str]:
|
|
mode = clean_text(mode) or "custom"
|
|
if mode == "target_only":
|
|
return unique_keep_order(targets)
|
|
if mode == "target_plus_reference":
|
|
return unique_keep_order([*targets, *refs])
|
|
if mode == "target_plus_comparative":
|
|
return unique_keep_order([*targets, *comps])
|
|
if mode == "target_plus_reference_plus_comparative":
|
|
return unique_keep_order([*targets, *refs, *comps])
|
|
return unique_keep_order([*targets, *refs, *comps])
|
|
|
|
|
|
def build_gene_family_scope(
|
|
input_dir: str | Path,
|
|
workspace: str | Path | None = None,
|
|
prefix: str = "",
|
|
family: str = "",
|
|
target_species: Iterable[str] | str | None = None,
|
|
reference_species: Iterable[str] | str | None = None,
|
|
comparative_species: Iterable[str] | str | None = None,
|
|
scope_mode: str = "custom",
|
|
) -> GeneFamilyScope:
|
|
input_path = Path(input_dir).resolve()
|
|
workspace_path = Path(workspace).resolve() if workspace else None
|
|
config_path = find_config_file(input_path)
|
|
rows: list[dict[str, Any]] = []
|
|
analysis: dict[str, str] = {}
|
|
warnings: list[str] = []
|
|
|
|
if config_path:
|
|
rows, analysis = read_config_workbook(config_path)
|
|
else:
|
|
warnings.append(f"No config.xlsx found in {input_path}")
|
|
|
|
family_value = infer_family(prefix, analysis, rows, family)
|
|
|
|
role_species: dict[str, list[str]] = {
|
|
ROLE_TARGET: [],
|
|
ROLE_REFERENCE: [],
|
|
ROLE_COMPARATIVE: [],
|
|
ROLE_SEED_REFERENCE: [],
|
|
}
|
|
scope_files: dict[str, dict[str, list[dict[str, Any]]]] = {}
|
|
|
|
for row in rows:
|
|
species = clean_text(row.get("Species_Code") or row.get("Species") or row.get("species"))
|
|
if not species:
|
|
continue
|
|
file_scope = clean_text(row.get("File_Scope"))
|
|
role = normalize_role(row.get("Species_Role"), row.get("Ref"), file_scope)
|
|
reference_use = normalize_reference_use(row.get("Reference_Use"))
|
|
if role:
|
|
role_species.setdefault(role, [])
|
|
if species not in role_species[role]:
|
|
role_species[role].append(species)
|
|
|
|
file_type = normalize_file_type(row.get("File_Type"), file_scope)
|
|
if not file_type:
|
|
continue
|
|
row_family = clean_text(row.get("Gene_Family"))
|
|
if row_family and family_value and row_family.lower() != family_value.lower() and file_type in {"family_pep", "family_cds"}:
|
|
continue
|
|
path = resolve_existing_path(input_path, workspace_path, species, family_value, row, file_type)
|
|
path_text, exists, size, mtime = file_record(path)
|
|
rec = ScopeFile(
|
|
original_file_name=clean_text(row.get("Original_File_Name")),
|
|
species=species,
|
|
file_type=file_type,
|
|
role=role,
|
|
file_scope=file_scope,
|
|
gene_family=row_family,
|
|
ref=parse_bool(row.get("Ref")),
|
|
reference_use=reference_use,
|
|
path=path_text,
|
|
exists=exists,
|
|
size=size,
|
|
mtime=mtime,
|
|
)
|
|
scope_files.setdefault(species, {}).setdefault(file_type, []).append(asdict(rec))
|
|
|
|
analysis_targets = split_csv(analysis.get("Target_Species"))
|
|
analysis_refs = split_csv(analysis.get("Reference_Species"))
|
|
analysis_primary_refs = split_csv(analysis.get("Primary_Reference_Species"))
|
|
analysis_identify_blast_refs = split_csv(analysis.get("Identify_Blast_Reference_Species"))
|
|
analysis_comps = split_csv(analysis.get("Comparative_Species"))
|
|
|
|
target_provided = target_species is not None
|
|
ref_provided = reference_species is not None
|
|
comp_provided = comparative_species is not None
|
|
|
|
target_list = split_csv(target_species) if isinstance(target_species, str) else unique_keep_order(target_species or [])
|
|
ref_list = split_csv(reference_species) if isinstance(reference_species, str) else unique_keep_order(reference_species or [])
|
|
comp_list = split_csv(comparative_species) if isinstance(comparative_species, str) else unique_keep_order(comparative_species or [])
|
|
|
|
if not target_provided and not target_list:
|
|
target_list = analysis_targets or role_species.get(ROLE_TARGET, [])
|
|
explicit_primary_refs = analysis_primary_refs or analysis_refs
|
|
if not ref_provided and not ref_list:
|
|
row_primary_refs = unique_keep_order(
|
|
row.get("Species_Code") or row.get("Species") or row.get("species")
|
|
for row in rows
|
|
if reference_use_includes(row.get("Reference_Use"), "downstream_primary")
|
|
)
|
|
ref_list = explicit_primary_refs or row_primary_refs or role_species.get(ROLE_REFERENCE, [])
|
|
identify_blast_ref_list = analysis_identify_blast_refs
|
|
if not identify_blast_ref_list:
|
|
identify_blast_ref_list = unique_keep_order(
|
|
row.get("Species_Code") or row.get("Species") or row.get("species")
|
|
for row in rows
|
|
if reference_use_includes(row.get("Reference_Use"), "identify_blast")
|
|
)
|
|
if not identify_blast_ref_list:
|
|
identify_blast_ref_list = unique_keep_order([*role_species.get(ROLE_SEED_REFERENCE, [])])
|
|
if analysis_identify_blast_refs and ref_list:
|
|
extra = [sp for sp in analysis_identify_blast_refs if sp not in ref_list]
|
|
if extra:
|
|
warnings.append("Identify_Blast_Reference_Species is used only for identify BLAST and will not be added to downstream reference_species: " + ",".join(extra))
|
|
if not analysis_primary_refs and analysis_refs:
|
|
warnings.append("Using legacy Analysis.Reference_Species as downstream primary reference; prefer Primary_Reference_Species for new projects.")
|
|
if not comp_provided and not comp_list:
|
|
comp_list = analysis_comps or role_species.get(ROLE_COMPARATIVE, [])
|
|
|
|
prefix_value = default_prefix(prefix, family_value, target_list)
|
|
selected = selected_species_for_mode(target_list, ref_list, comp_list, scope_mode)
|
|
scope_id = build_scope_id(family_value, target_list, ref_list, comp_list, clean_text(scope_mode) or "custom")
|
|
|
|
return GeneFamilyScope(
|
|
family=family_value,
|
|
prefix=prefix_value,
|
|
scope_id=scope_id,
|
|
scope_mode=clean_text(scope_mode) or "custom",
|
|
target_species=target_list,
|
|
reference_species=ref_list,
|
|
primary_reference_species=ref_list,
|
|
identify_blast_reference_species=identify_blast_ref_list,
|
|
comparative_species=comp_list,
|
|
selected_species=selected,
|
|
config_path=str(config_path.resolve()) if config_path else "",
|
|
files=scope_files,
|
|
analysis=analysis,
|
|
warnings=warnings,
|
|
)
|
|
|
|
|
|
def write_scope(scope: GeneFamilyScope, output_path: str | Path) -> Path:
|
|
path = Path(output_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(scope.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def read_scope(scope_path: str | Path) -> dict[str, Any]:
|
|
return json.loads(Path(scope_path).read_text(encoding="utf-8"))
|
|
|
|
|
|
def scope_manifest_path(workspace: str | Path, prefix: str, scope_id: str) -> Path:
|
|
return Path(workspace) / "final" / f"{sanitize_token(prefix)}.{sanitize_token(scope_id)}.scope.json"
|
|
|
|
|
|
def sha256_file(path: str | Path, chunk_size: int = 1024 * 1024) -> str:
|
|
hasher = hashlib.sha256()
|
|
with Path(path).open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(chunk_size), b""):
|
|
hasher.update(chunk)
|
|
return hasher.hexdigest()
|