441 lines
15 KiB
Python
441 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from gyxx_flow.security import scan_repository
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
MODULE_ROOT = (
|
|
PROJECT_ROOT
|
|
/ "src"
|
|
/ "gyxx_flow"
|
|
/ "modules"
|
|
/ "product_commerce"
|
|
)
|
|
MIGRATED_TEST_ROOT = PROJECT_ROOT / "tests" / "modules" / "product_commerce"
|
|
HISTORY_ROOT = PROJECT_ROOT / "docs" / "history" / "product-commerce"
|
|
MANIFEST_PATH = (
|
|
PROJECT_ROOT / "config" / "source-manifests" / "product_commerce.json"
|
|
)
|
|
|
|
PRODUCTION_PYTHON = {
|
|
"aggregate_daily_final.py",
|
|
"analyze_style.py",
|
|
"backfill_poseidon_sales.py",
|
|
"backfill_collect.py",
|
|
"backfill_one_day.py",
|
|
"check_nine_day_decline.py",
|
|
"collect_dy_market_rank.py",
|
|
"collect_dy_persona_to_bitable.py",
|
|
"collect_erp_yesterday_metrics.py",
|
|
"collect_jd_market_rank.py",
|
|
"collect_jd_persona_to_bitable.py",
|
|
"collect_persona_to_bitable.py",
|
|
"collect_retry_utils.py",
|
|
"collect_sycm_market_rank.py",
|
|
"config/__init__.py",
|
|
"config/style_config_loader.py",
|
|
"db/__init__.py",
|
|
"db/sync_dim_style.py",
|
|
"db/sync_sku_master.py",
|
|
"dy_audience_profile_collect.py",
|
|
"dy_product_scraping.py",
|
|
"erp_login_product_analysis.py",
|
|
"erp_metric_overrides.py",
|
|
"export_bitable_records.py",
|
|
"feishu_doc_native.py",
|
|
"import_product_daily.py",
|
|
"import_product_reviews.py",
|
|
"insert_bitable_records.py",
|
|
"jd_main_image_collector.py",
|
|
"jd_product_data_collector.py",
|
|
"jd_self_inventory_sales_collector.py",
|
|
"lark_cli_runtime.py",
|
|
"main_image_db.py",
|
|
"main_image_paths.py",
|
|
"market_rank_hermes_notification.py",
|
|
"market_rank_limits.py",
|
|
"market_rank_product_import.py",
|
|
"market_rank_report_archive.py",
|
|
"orchestrate_daily_collection.py",
|
|
"orchestrate_market_rank_collection.py",
|
|
"orchestrate_review_collection.py",
|
|
"reapply_erp_override.py",
|
|
"rebuild_market_rank_documents.py",
|
|
"run_alerts_with_retry.py",
|
|
"run_daily_persona.py",
|
|
"run_weekly_jd_main_image.py",
|
|
"run_weekly_main_image.py",
|
|
"scripts/insert_jd_main_image_records.py",
|
|
"scripts/insert_main_image_records.py",
|
|
"taobao_dmp_item_crowd_insight_screenshots.py",
|
|
"taobao_sycm_collect.py",
|
|
"taobao_sycm_collect_backfill.py",
|
|
"taobao_sycm_products.py",
|
|
"taobao_wanxiang_ai_creative_report.py",
|
|
"upload_video_to_guanghe.py",
|
|
"vendors/dy-data-flow/adaptive_selectors.py",
|
|
"vendors/dy-data-flow/dy_store_competitor_store_scraping.py",
|
|
"vendors/dy-data-flow/dynamic_session_src.py",
|
|
"vendors/jd-data-flow/collection_progress.py",
|
|
"vendors/jd-data-flow/config.py",
|
|
"vendors/jd-data-flow/jd_data_collector.py",
|
|
"vendors/jd-data-flow/jd_peer_product_data_collector.py",
|
|
"vendors/jd-data-flow/jd_product_data_collector.py",
|
|
"vendors/jd-data-flow/state.py",
|
|
"weekly_aggregate.py",
|
|
}
|
|
|
|
RESOURCES = {
|
|
".env.example",
|
|
"bitable_main_image_map.json",
|
|
"bitable_style_map.json",
|
|
"config/auto-flow-config.example.json",
|
|
"db/schema.sql",
|
|
"styles_input.json",
|
|
}
|
|
|
|
LEGACY_TESTS = {
|
|
"test_backfill_poseidon_sales.py",
|
|
"test_db_config.py",
|
|
"test_dy_market_rank.py",
|
|
"test_dy_session_reuse.py",
|
|
"test_erp_metric_overrides.py",
|
|
"test_erp_no_data_freshness.py",
|
|
"test_erp_slow_skip_codes.py",
|
|
"test_feishu_doc_native.py",
|
|
"test_guanghe_metadata.py",
|
|
"test_guanghe_store_routing.py",
|
|
"test_import_product_daily.py",
|
|
"test_jd_enter_shangzhi.py",
|
|
"test_jd_market_rank.py",
|
|
"test_main_image_concurrency.py",
|
|
"test_market_rank_hermes_notification.py",
|
|
"test_market_rank_limits.py",
|
|
"test_market_rank_product_import.py",
|
|
"test_market_rank_report_archive.py",
|
|
"test_market_rank_workflow.py",
|
|
"test_persona_launcher.py",
|
|
"test_style_analysis_orchestration.py",
|
|
"test_style_config_loader_erp_source.py",
|
|
"test_sycm_market_rank.py",
|
|
"test_tm_persona_recovery.py",
|
|
"test_wanxiang_report_template.py",
|
|
}
|
|
|
|
LAUNCHER_REFERENCES = {
|
|
"run_alerts.bat",
|
|
"run_daily_collect.bat",
|
|
"run_daily_import.bat",
|
|
"run_daily_import.ps1",
|
|
"run_daily_persona.bat",
|
|
"run_style_analysis_3d.bat",
|
|
"run_weekly_jd_main_image.bat",
|
|
"run_weekly_main_image.bat",
|
|
"run_weekly_market_rank.bat",
|
|
}
|
|
|
|
INTENTIONALLY_EXCLUDED = {
|
|
"debug_canvax_dump.py",
|
|
"debug_canvax_dump2.py",
|
|
"debug_erp_filter.py",
|
|
"debug_jd_main_image_snapshot.py",
|
|
"inspect_dy_comment_filters.py",
|
|
"jd_collect_test.py",
|
|
"scripts/inspect_fields.py",
|
|
}
|
|
|
|
SUBPROCESS_TARGETS = {
|
|
"aggregate_daily_final.py",
|
|
"analyze_style.py",
|
|
"check_nine_day_decline.py",
|
|
"collect_dy_market_rank.py",
|
|
"collect_dy_persona_to_bitable.py",
|
|
"collect_erp_yesterday_metrics.py",
|
|
"collect_jd_market_rank.py",
|
|
"collect_jd_persona_to_bitable.py",
|
|
"collect_persona_to_bitable.py",
|
|
"collect_sycm_market_rank.py",
|
|
"dy_audience_profile_collect.py",
|
|
"dy_product_scraping.py",
|
|
"export_bitable_records.py",
|
|
"import_product_reviews.py",
|
|
"insert_bitable_records.py",
|
|
"jd_main_image_collector.py",
|
|
"jd_product_data_collector.py",
|
|
"orchestrate_daily_collection.py",
|
|
"scripts/insert_jd_main_image_records.py",
|
|
"scripts/insert_main_image_records.py",
|
|
"taobao_dmp_item_crowd_insight_screenshots.py",
|
|
"taobao_sycm_collect.py",
|
|
"taobao_sycm_collect_backfill.py",
|
|
"taobao_sycm_products.py",
|
|
"taobao_wanxiang_ai_creative_report.py",
|
|
}
|
|
|
|
|
|
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 _load_runtime_paths():
|
|
path = MODULE_ROOT / "runtime_paths.py"
|
|
spec = importlib.util.spec_from_file_location("product_runtime_paths_test", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_product_runtime_contains_the_complete_reviewed_source_snapshot() -> None:
|
|
assert len(PRODUCTION_PYTHON) == 65
|
|
assert all((MODULE_ROOT / relative).is_file() for relative in PRODUCTION_PYTHON)
|
|
assert all((MODULE_ROOT / relative).is_file() for relative in RESOURCES)
|
|
assert all((MIGRATED_TEST_ROOT / name).is_file() for name in LEGACY_TESTS)
|
|
assert all(
|
|
(HISTORY_ROOT / "launchers_reference" / name).is_file()
|
|
for name in LAUNCHER_REFERENCES
|
|
)
|
|
assert (HISTORY_ROOT / "MIGRATION_STATUS.md").is_file()
|
|
|
|
|
|
def test_product_source_manifest_is_complete_and_verifiable() -> None:
|
|
manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
|
|
assert set(manifest) == {
|
|
"schema_version",
|
|
"module",
|
|
"snapshot",
|
|
"files",
|
|
"intentionally_excluded",
|
|
"transformation_rules",
|
|
}
|
|
assert manifest["schema_version"] == 1
|
|
assert manifest["module"] == "product_commerce"
|
|
assert manifest["snapshot"] == "current-filesystem"
|
|
audit = next(
|
|
rule["metadata"]
|
|
for rule in manifest["transformation_rules"]
|
|
if rule["id"] == "migration-audit"
|
|
)
|
|
assert audit["source_project"] == "product-collector-analyze-flow"
|
|
assert audit["snapshot_kind"] == "working_tree"
|
|
security_review = audit["security_review"]
|
|
assert security_review["reviewed_source_python_findings"] == 15
|
|
assert security_review["migrated_findings"] == 14
|
|
assert security_review["excluded_findings"] == 1
|
|
assert security_review["target_findings"] == 0
|
|
assert len(security_review["decisions"]) == 15
|
|
assert {
|
|
item["source_relative_path"]
|
|
for item in manifest["intentionally_excluded"]
|
|
if "source_relative_path" in item
|
|
} == INTENTIONALLY_EXCLUDED
|
|
assert all(
|
|
item.get("source_relative_path") or item.get("pattern")
|
|
for item in manifest["intentionally_excluded"]
|
|
)
|
|
|
|
expected_targets = {
|
|
*(
|
|
f"src/gyxx_flow/modules/product_commerce/{path}"
|
|
for path in PRODUCTION_PYTHON | RESOURCES
|
|
),
|
|
*(
|
|
f"tests/modules/product_commerce/{name}"
|
|
for name in LEGACY_TESTS
|
|
),
|
|
*(
|
|
f"docs/history/product-commerce/launchers_reference/{name}"
|
|
for name in LAUNCHER_REFERENCES
|
|
),
|
|
}
|
|
entries = manifest["files"]
|
|
assert {item["target_relative_path"] for item in entries} == expected_targets
|
|
assert len(entries) == len(expected_targets)
|
|
|
|
for item in entries:
|
|
target = PROJECT_ROOT / item["target_relative_path"]
|
|
assert target.is_file()
|
|
assert item["category"] in {
|
|
"production_source",
|
|
"source_resource",
|
|
"runtime_resource",
|
|
"regression_test",
|
|
"launcher_provenance",
|
|
}
|
|
assert isinstance(item["transformed"], bool)
|
|
assert len(item["source_sha256"]) == 64
|
|
assert item["target_sha256"] == _sha256(target)
|
|
if not item["transformed"]:
|
|
assert item["source_sha256"] == item["target_sha256"]
|
|
|
|
generated = audit["generated_files"]
|
|
assert {item["target_relative_path"] for item in generated} == {
|
|
"src/gyxx_flow/modules/product_commerce/runtime_paths.py"
|
|
}
|
|
for item in generated:
|
|
target = PROJECT_ROOT / item["target_relative_path"]
|
|
assert item["target_sha256"] == _sha256(target)
|
|
assert manifest["transformation_rules"]
|
|
|
|
|
|
def test_migrated_product_snapshot_has_no_plaintext_credentials_or_old_roots() -> None:
|
|
assert scan_repository(MODULE_ROOT) == []
|
|
|
|
forbidden = ("E:\\auto-flow", "D:\\product-collector-analyze-flow")
|
|
for path in MODULE_ROOT.rglob("*"):
|
|
if not path.is_file() or path.suffix.casefold() in {".pyc", ".pyo"}:
|
|
continue
|
|
try:
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
except (OSError, UnicodeDecodeError):
|
|
continue
|
|
assert not any(value.casefold() in text.casefold() for value in forbidden), path
|
|
|
|
|
|
def test_product_runtime_paths_are_portable_and_layered(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("GYXX_MODULE_ROOT", str(MODULE_ROOT))
|
|
monkeypatch.setenv("GYXX_DATA_ROOT", str(tmp_path))
|
|
paths = _load_runtime_paths()
|
|
|
|
assert paths.MODULE_ROOT == MODULE_ROOT.resolve()
|
|
assert paths.DATA_HOME == tmp_path.resolve()
|
|
assert paths.RAW_DATA_ROOT == tmp_path / "data" / "raw" / "product_commerce"
|
|
assert paths.NORMALIZED_DATA_ROOT == (
|
|
tmp_path / "data" / "normalized" / "product_commerce"
|
|
)
|
|
assert paths.CURATED_DATA_ROOT == (
|
|
tmp_path / "data" / "curated" / "product_commerce"
|
|
)
|
|
assert paths.EXPORTS_DATA_ROOT == (
|
|
tmp_path / "data" / "exports" / "product_commerce"
|
|
)
|
|
assert paths.STATE_ROOT == tmp_path / "state" / "product_commerce"
|
|
assert paths.PROFILE_ROOT == (
|
|
tmp_path / "state" / "product_commerce" / "browser-profiles"
|
|
)
|
|
assert paths.LOG_ROOT == tmp_path / "logs" / "product_commerce"
|
|
assert paths.TMP_ROOT == tmp_path / "tmp" / "product_commerce"
|
|
|
|
assert paths.vendor_source_root("jd") == (
|
|
MODULE_ROOT / "vendors" / "jd-data-flow"
|
|
).resolve()
|
|
assert paths.vendor_source_root("dy") == (
|
|
MODULE_ROOT / "vendors" / "dy-data-flow"
|
|
).resolve()
|
|
for relative in SUBPROCESS_TARGETS:
|
|
assert paths.runtime_script(relative).is_file()
|
|
|
|
|
|
def test_product_runtime_has_no_machine_or_legacy_path_escape_hatches() -> None:
|
|
forbidden = (
|
|
"D:\\yingxiaoyunying",
|
|
"D:\\shop-data-flow",
|
|
"D:\\product-collector-analyze-flow",
|
|
"E:\\auto-flow",
|
|
"D:\\jd-data-flow",
|
|
"E:\\jd-data-flow",
|
|
"D:\\dy-data-flow",
|
|
"E:\\dy-data-flow",
|
|
"JD_DATA_FLOW_ROOT",
|
|
"DY_DATA_FLOW_ROOT",
|
|
"AUTO_FLOW_CONFIG",
|
|
"C:\\ChromeDebug",
|
|
"C:\\Users\\Administrator",
|
|
'PROJECT_ROOT / "data"',
|
|
'ROOT / "data"',
|
|
'os.path.join(PROJECT_ROOT, "data"',
|
|
)
|
|
for path in MODULE_ROOT.rglob("*"):
|
|
if not path.is_file() or path.suffix.casefold() in {".pyc", ".pyo"}:
|
|
continue
|
|
try:
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
except (OSError, UnicodeDecodeError):
|
|
continue
|
|
assert not any(value.casefold() in text.casefold() for value in forbidden), path
|
|
|
|
|
|
def test_product_runtime_has_no_hardcoded_long_hex_keys() -> None:
|
|
hardcoded_key = re.compile(
|
|
r"(?is)(?:api[_-]?key|secret|token).{0,160}?[\"'][0-9a-f]{32,}[\"']"
|
|
)
|
|
for path in MODULE_ROOT.rglob("*.py"):
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
assert hardcoded_key.search(text) is None, path
|
|
|
|
analyzer = (MODULE_ROOT / "analyze_style.py").read_text(
|
|
encoding="utf-8-sig"
|
|
)
|
|
assert "call_direct_llm" in analyzer
|
|
assert "STYLE_ANALYSIS_LLM_MODEL" in analyzer
|
|
assert "analyze_style_with_hermes" not in analyzer
|
|
|
|
|
|
def test_product_core_runtime_has_no_known_f821_regressions() -> None:
|
|
erp_tree = ast.parse(
|
|
(MODULE_ROOT / "collect_erp_yesterday_metrics.py").read_text(
|
|
encoding="utf-8-sig"
|
|
)
|
|
)
|
|
assert any(
|
|
isinstance(node, ast.FunctionDef) and node.name == "_ensure_iframe_alive"
|
|
for node in erp_tree.body
|
|
)
|
|
|
|
class ScopeNames(ast.NodeVisitor):
|
|
def __init__(self, root: ast.FunctionDef) -> None:
|
|
self.root = root
|
|
self.loaded: set[str] = set()
|
|
self.stored: set[str] = set()
|
|
|
|
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
|
|
if node is self.root:
|
|
self.generic_visit(node)
|
|
|
|
def visit_Name(self, node: ast.Name) -> None:
|
|
if isinstance(node.ctx, ast.Load):
|
|
self.loaded.add(node.id)
|
|
elif isinstance(node.ctx, ast.Store):
|
|
self.stored.add(node.id)
|
|
|
|
jd_paths = (
|
|
MODULE_ROOT / "jd_product_data_collector.py",
|
|
MODULE_ROOT / "vendors" / "jd-data-flow" / "jd_product_data_collector.py",
|
|
)
|
|
for path in jd_paths:
|
|
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
|
|
implementations = [
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "collect_each_spu_detail"
|
|
]
|
|
assert implementations
|
|
for implementation in implementations:
|
|
names = ScopeNames(implementation)
|
|
names.visit(implementation)
|
|
assert "week_range" not in names.loaded or "week_range" in names.stored
|
|
|
|
|
|
def test_dynamic_session_snapshot_is_classified_as_source_resource() -> None:
|
|
manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
|
|
item = next(
|
|
item
|
|
for item in manifest["files"]
|
|
if item["source_relative_path"]
|
|
== "vendors/dy-data-flow/dynamic_session_src.py"
|
|
)
|
|
assert item["category"] == "source_resource"
|
|
assert item["lint_policy"] == "not_a_standalone_module"
|
|
assert item["reason"]
|