215 lines
7.5 KiB
Python
215 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
from gyxx_flow.security.scanner import scan_repository
|
|
|
|
PROJECT_ROOT = Path(__file__).parents[1]
|
|
MODULE_ROOT = (
|
|
PROJECT_ROOT / "src" / "gyxx_flow" / "modules" / "content_marketing"
|
|
)
|
|
MANIFEST_PATH = (
|
|
PROJECT_ROOT / "config" / "source-manifests" / "content_marketing.json"
|
|
)
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
OLD_PROJECT_ROOTS = (
|
|
"d:\\yingxiaoyunying",
|
|
"d:/yingxiaoyunying",
|
|
"d:\\shop-data-flow",
|
|
"d:/shop-data-flow",
|
|
"d:\\product-collector-analyze-flow",
|
|
"d:/product-collector-analyze-flow",
|
|
"e:\\auto-flow",
|
|
"e:/auto-flow",
|
|
)
|
|
FIXED_USER_PATHS = (
|
|
"c:\\users\\administrator",
|
|
"c:/users/administrator",
|
|
)
|
|
EXPECTED_CATEGORIES = {
|
|
"config",
|
|
"dependency",
|
|
"source",
|
|
"sql",
|
|
"tool",
|
|
}
|
|
REQUIRED_SOURCES = {
|
|
"run_all.py",
|
|
"bilibili_scraper.py",
|
|
"pgy_xhs_scraper_v2.py",
|
|
"xingtu_scraper_v2.py",
|
|
"chanmama_scraper.py",
|
|
"daily_creator_exposure_scope.py",
|
|
"daily_marketing_report.py",
|
|
"data/tools/friday_relogin_parallel.py",
|
|
"data/tools/schema_gyxx_super_data.sql",
|
|
"data/tools/migrations/005_style_product_profile.sql",
|
|
"data/config/db.env.example",
|
|
"data/config/款式_多维表格_对照.json",
|
|
}
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _manifest() -> dict[str, object]:
|
|
return json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_content_source_snapshot_manifest_is_complete_and_well_formed() -> None:
|
|
manifest = _manifest()
|
|
|
|
assert manifest["schema_version"] == 1
|
|
assert manifest["module"] == "content_marketing"
|
|
files = manifest["files"]
|
|
assert isinstance(files, list)
|
|
assert len(files) == 66
|
|
|
|
sources = [entry["source_relative_path"] for entry in files]
|
|
targets = [entry["target_relative_path"] for entry in files]
|
|
assert sources == sorted(sources, key=str.casefold)
|
|
assert len(sources) == len(set(sources))
|
|
assert len(targets) == len(set(targets))
|
|
assert REQUIRED_SOURCES <= set(sources)
|
|
assert {entry["category"] for entry in files} == EXPECTED_CATEGORIES
|
|
|
|
for entry in files:
|
|
relative = PurePosixPath(entry["target_relative_path"])
|
|
assert not relative.is_absolute()
|
|
assert ".." not in relative.parts
|
|
assert relative.parts[:4] == (
|
|
"src",
|
|
"gyxx_flow",
|
|
"modules",
|
|
"content_marketing",
|
|
)
|
|
assert relative.parts[4] != "runtime"
|
|
assert SHA256.fullmatch(entry["source_sha256"])
|
|
assert SHA256.fullmatch(entry["target_sha256"])
|
|
assert entry["transformed"] is (
|
|
entry["source_sha256"] != entry["target_sha256"]
|
|
)
|
|
|
|
|
|
def test_content_source_snapshot_targets_exist_and_match_manifest_hashes() -> None:
|
|
for entry in _manifest()["files"]:
|
|
target = PROJECT_ROOT.joinpath(*PurePosixPath(entry["target_relative_path"]).parts)
|
|
assert target.is_file(), entry["target_relative_path"]
|
|
assert _sha256(target) == entry["target_sha256"]
|
|
|
|
generated = _manifest()["generated_files"]
|
|
assert [entry["target_relative_path"] for entry in generated] == [
|
|
"src/gyxx_flow/modules/content_marketing/runtime_paths.py"
|
|
]
|
|
for entry in generated:
|
|
target = PROJECT_ROOT.joinpath(*PurePosixPath(entry["target_relative_path"]).parts)
|
|
assert target.is_file(), entry["target_relative_path"]
|
|
assert SHA256.fullmatch(entry["target_sha256"])
|
|
assert _sha256(target) == entry["target_sha256"]
|
|
|
|
|
|
def test_content_mapping_snapshot_includes_latest_source_styles() -> None:
|
|
config_root = MODULE_ROOT / "data" / "config"
|
|
collaborator = json.loads(
|
|
(config_root / "款式_多维表格_对照.json").read_text(encoding="utf-8")
|
|
)
|
|
self_operated = json.loads(
|
|
(config_root / "款式_多维表格_对照_自营.json").read_text(encoding="utf-8")
|
|
)
|
|
|
|
collaborator_by_name = {item["name"]: item for item in collaborator["tables"]}
|
|
self_by_name = {item["name"]: item for item in self_operated["tables"]}
|
|
assert collaborator_by_name["极星2"]["field_map"]["note_title"] == {
|
|
"field_id": "fldF40Tdh4",
|
|
"field_name": "发布笔记标题",
|
|
}
|
|
assert collaborator_by_name["极星托特"]["table_id"] == "tbl6L5fNUNsDGGcN"
|
|
assert self_by_name["凌云3"]["table_id"] == "tblMCT9eDe9QGTP2"
|
|
assert self_by_name["极星双肩2"]["table_id"] == "tblzVYspLYZHrKMD"
|
|
|
|
|
|
def test_content_source_snapshot_exclusions_cover_runtime_and_sensitive_state() -> None:
|
|
excluded = {entry["pattern"] for entry in _manifest()["intentionally_excluded"]}
|
|
|
|
assert {
|
|
".git/**",
|
|
"**/__pycache__/**",
|
|
"data/config/*.env",
|
|
"data/**/*cookies*",
|
|
".*chrome_profile*/**",
|
|
"data/logs/**",
|
|
"data/v2_results/**",
|
|
"data/notes/**",
|
|
"data/reports/**",
|
|
"data/tmp/**",
|
|
"tests/**",
|
|
"**/*.bat",
|
|
"**/*.ps1",
|
|
} <= excluded
|
|
|
|
|
|
def test_content_source_snapshot_has_no_legacy_roots_or_plaintext_credentials() -> None:
|
|
manifest_text = MANIFEST_PATH.read_text(encoding="utf-8").casefold()
|
|
assert not any(root in manifest_text for root in OLD_PROJECT_ROOTS)
|
|
|
|
for path in MODULE_ROOT.rglob("*"):
|
|
if not path.is_file() or b"\x00" in path.read_bytes():
|
|
continue
|
|
text = path.read_text(encoding="utf-8-sig", errors="strict").casefold()
|
|
assert not any(root in text for root in OLD_PROJECT_ROOTS), path
|
|
assert not any(root in text for root in FIXED_USER_PATHS), path
|
|
|
|
assert scan_repository(MODULE_ROOT) == []
|
|
|
|
|
|
def test_chanmama_credentials_are_externalized_in_the_copied_source() -> None:
|
|
source = (MODULE_ROOT / "chanmama_scraper.py").read_text(encoding="utf-8-sig")
|
|
|
|
assert 'os.getenv("CHANMAMA_ACCOUNT", "")' in source
|
|
assert 'os.getenv("CHANMAMA_PASSWORD", "")' in source
|
|
assert not re.search(r'^PASSWORD\s*=\s*["\'][^"\']+["\']', source, re.MULTILINE)
|
|
|
|
|
|
def test_content_module_has_one_portable_path_boundary() -> None:
|
|
python_sources = list(MODULE_ROOT.rglob("*.py"))
|
|
path_mutations: list[Path] = []
|
|
forbidden_cross_module_imports: list[Path] = []
|
|
|
|
for path in python_sources:
|
|
source = path.read_text(encoding="utf-8-sig")
|
|
if re.search(r"sys\.path\.(?:insert|append)\(", source):
|
|
path_mutations.append(path.relative_to(MODULE_ROOT))
|
|
if "lark_cli_runtime" in source:
|
|
forbidden_cross_module_imports.append(path.relative_to(MODULE_ROOT))
|
|
|
|
assert path_mutations == []
|
|
assert forbidden_cross_module_imports == []
|
|
|
|
runtime_paths = (
|
|
MODULE_ROOT / "runtime_paths.py"
|
|
).read_text(encoding="utf-8-sig")
|
|
assert "GYXX_DATA_ROOT" in runtime_paths
|
|
assert "GYXX_MODULE_ROOT" in runtime_paths
|
|
|
|
|
|
def test_content_module_uses_python_commands_and_portable_data_paths() -> None:
|
|
assert list(MODULE_ROOT.rglob("*.bat")) == []
|
|
assert list(MODULE_ROOT.rglob("*.ps1")) == []
|
|
|
|
path_constants = (MODULE_ROOT / "runtime_paths.py").read_text(
|
|
encoding="utf-8-sig"
|
|
)
|
|
assert 'DataLayout(data_root).for_module("content_marketing")' in path_constants
|
|
assert "raw_root = _DATA_PATHS.raw_root" in path_constants
|
|
assert "normalized_root = _DATA_PATHS.normalized_root" in path_constants
|
|
assert "curated_root = _DATA_PATHS.curated_root" in path_constants
|
|
assert "exports_root = _DATA_PATHS.exports_root" in path_constants
|