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"]