wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(crawler): read back pixelrag visual evidence

Browse Source
This commit is contained in:
ogt
2026-07-09 17:39:33 +08:00
parent 14a8a2052d
commit 1a435b4c78
4 changed files with 286 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/guides/browse_sh_crawler_playbook.md
View File

@@ -85,6 +85,7 @@ python scripts/ops/report_pixelrag_crawler_integration.py \
```text
/api/ai-automation/pixelrag-crawler-integration?platform=momo
/api/ai-automation/pixelrag-crawler-integration?platform=pchome&manifest_url=https://24h.pchome.com.tw/prod/TEST-000000001&crawler=PChomeCrawler.search_products&trigger_reason=parser_empty
/api/ai-automation/pixelrag-visual-evidence-readback?platform=pchome&manifest_id=4a93e95e5afb414bc8c3
```
視覺證據 capture worker：
@@ -115,6 +116,14 @@ python3 scripts/ops/capture_pixelrag_visual_evidence.py \
--max-tiles 12
```
capture 成功後，receipt 會落在：
```text
runtime_artifacts/pixelrag_visual_evidence/<platform>/<manifest_id>/capture_receipt.json
```
API readback 只讀 receipt 和檔案存在狀態，不重新抓外站、不寫資料庫。
安全邊界：
- read-only；不登入、不下單、不加入購物車、不寫第三方狀態。

14
routes/system_public_routes.py
View File

@@ -659,6 +659,20 @@ def ai_automation_pixelrag_crawler_integration_api():
))
@system_public_bp.route('/api/ai-automation/pixelrag-visual-evidence-readback')
@login_required
def ai_automation_pixelrag_visual_evidence_readback_api():
"""Read-only PixelRAG visual evidence capture artifact readback."""
from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback
max_age_hours = request.args.get('max_age_hours', 168, type=int)
return jsonify(build_pixelrag_visual_evidence_readback(
platform=str(request.args.get('platform') or '').strip() or None,
manifest_id=str(request.args.get('manifest_id') or '').strip() or None,
max_age_hours=max(1, min(max_age_hours or 168, 720)),
))
@system_public_bp.route('/api/ai-automation/smoke/history/export')
@login_required
def ai_automation_smoke_history_export():

161
services/pixelrag_crawler_integration_service.py
View File

@@ -8,15 +8,19 @@ integration assessment for the existing MOMO/PChome crawler stack.
from __future__ import annotations
import hashlib
import json
import math
import os
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping
from urllib.parse import urlparse
POLICY = "read_only_pixelrag_crawler_integration_assessment_v1"
MANIFEST_POLICY = "read_only_pixelrag_visual_evidence_manifest_v1"
READBACK_POLICY = "read_only_pixelrag_visual_evidence_readback_v1"
RESULT_PHASE1_READY = "PIXELRAG_VISUAL_EVIDENCE_PHASE1_READY"
RESULT_BLOCKED_NO_CAPTURE = "PIXELRAG_BLOCKED_NO_VISUAL_CAPTURE"
RESULT_MANIFEST_READY = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY"
@@ -25,6 +29,11 @@ RESULT_MANIFEST_REJECTED = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_REJECTED"
ALLOWED_PLATFORMS = ("momo", "pchome", "market_intel", "external_market")
DEFAULT_VIEWPORT = {"name": "desktop-1440", "width": 1440, "height": 950}
DEFAULT_TILE_SIZE = {"width": 512, "height": 512}
DEFAULT_ARTIFACT_ROOT = os.getenv(
"PIXELRAG_VISUAL_EVIDENCE_ROOT",
"runtime_artifacts/pixelrag_visual_evidence",
)
DEFAULT_ARTIFACT_MAX_AGE_HOURS = int(os.getenv("PIXELRAG_VISUAL_EVIDENCE_MAX_AGE_HOURS", "168"))
VISUAL_FALLBACK_CONFIDENCE_TRIGGERS = {
"low",
"manual_review",
@@ -156,6 +165,25 @@ def _controlled_apply_boundary() -> dict[str, Any]:
}
def _parse_iso_datetime(value: Any) -> datetime | None:
if not value:
return None
try:
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _safe_int(value: Any, default: int = 0) -> int:
try:
return int(value or default)
except (TypeError, ValueError):
return default
def _build_phases(capabilities: Mapping[str, bool]) -> list[dict[str, Any]]:
capture_ready = bool(capabilities["playwright_artifact_pipeline"])
structured_ready = bool(capabilities["structured_crawler_api"])
@@ -490,14 +518,147 @@ def build_pixelrag_crawler_integration_assessment(
return payload
def _receipt_candidates(
root: Path,
*,
platform: str | None = None,
manifest_id: str | None = None,
) -> list[Path]:
if manifest_id and platform:
candidate = root / platform / manifest_id / "capture_receipt.json"
return [candidate] if candidate.exists() else []
pattern = "*/capture_receipt.json" if platform and manifest_id else "*/**/capture_receipt.json"
search_root = root / platform if platform and not manifest_id else root
if not search_root.exists():
return []
return sorted(search_root.glob(pattern), key=lambda path: path.stat().st_mtime, reverse=True)
def build_pixelrag_visual_evidence_readback(
*,
artifact_root: str | Path | None = None,
platform: str | None = None,
manifest_id: str | None = None,
max_age_hours: int | None = None,
) -> dict[str, Any]:
"""Read latest PixelRAG capture receipt and file presence without running capture."""
root = Path(artifact_root or DEFAULT_ARTIFACT_ROOT)
max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
platform_filter = str(platform or "").strip().lower() or None
manifest_filter = str(manifest_id or "").strip() or None
candidates = _receipt_candidates(root, platform=platform_filter, manifest_id=manifest_filter)
now = datetime.now(timezone.utc)
if not candidates:
return {
"success": False,
"policy": READBACK_POLICY,
"status": "warning",
"artifact_root": str(root),
"platform": platform_filter,
"manifest_id": manifest_filter,
"summary": {
"receipt_count": 0,
"tile_file_count": 0,
"missing_file_count": 0,
"age_hours": None,
"stale": True,
},
"latest_receipt": None,
"next_machine_action": "run_pixelrag_visual_capture_worker",
"controlled_apply": {
"network_call": False,
"db_write": False,
"secret_read": False,
"production_price_write": False,
},
}
latest_path = candidates[0]
errors: list[str] = []
try:
receipt = json.loads(latest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
receipt = {}
errors.append(str(exc)[:300])
generated_at = _parse_iso_datetime(receipt.get("generated_at"))
age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None
stale = age_hours is None or age_hours > max_age
files = list(receipt.get("files") or [])
file_statuses = []
missing_file_count = 0
tile_file_count = 0
for item in files:
path = Path(str(item.get("path") or ""))
exists = path.exists()
if not exists:
missing_file_count += 1
if item.get("kind") == "tile":
tile_file_count += 1
file_statuses.append({
"kind": item.get("kind"),
"path": str(path),
"exists": exists,
})
capture_target = receipt.get("capture_target") or {}
tile_plan = receipt.get("tile_plan") or {}
status = "ok" if receipt.get("status") == "captured" and missing_file_count == 0 and not stale else "warning"
if errors:
status = "critical"
return {
"success": not errors,
"policy": READBACK_POLICY,
"status": status,
"artifact_root": str(root),
"platform": platform_filter,
"manifest_id": manifest_filter or receipt.get("manifest_id"),
"summary": {
"receipt_count": len(candidates),
"tile_file_count": tile_file_count,
"missing_file_count": missing_file_count,
"planned_tile_count": _safe_int(tile_plan.get("planned_tile_count")),
"emitted_tile_count": _safe_int(tile_plan.get("emitted_tile_count")),
"http_status": _safe_int(receipt.get("http_status")),
"age_hours": round(age_hours, 3) if age_hours is not None else None,
"stale": stale,
"errors": errors,
},
"latest_receipt": {
"path": str(latest_path),
"generated_at": receipt.get("generated_at"),
"capture_status": receipt.get("status"),
"capture_target": capture_target,
"page_metrics": receipt.get("page_metrics") or {},
"tile_plan": tile_plan,
"files": file_statuses,
},
"next_machine_action": (
"keep_pixelrag_visual_evidence_receipt"
if status == "ok"
else "run_pixelrag_visual_capture_worker"
),
"controlled_apply": {
"network_call": False,
"db_write": False,
"secret_read": False,
"production_price_write": False,
},
}
__all__ = [
"MANIFEST_POLICY",
"POLICY",
"READBACK_POLICY",
"RESULT_PHASE1_READY",
"RESULT_BLOCKED_NO_CAPTURE",
"RESULT_MANIFEST_READY",
"RESULT_MANIFEST_REJECTED",
"build_pixelrag_crawler_integration_assessment",
"build_pixelrag_visual_evidence_readback",
"build_pixelrag_visual_evidence_manifest",
"should_emit_visual_evidence_fallback",
]

102
tests/test_pixelrag_crawler_integration_service.py
View File

@@ -2,6 +2,7 @@ import json
import shutil
import subprocess
import sys
from datetime import datetime, timezone
import pytest
@@ -234,6 +235,107 @@ def test_pixelrag_ai_automation_route_returns_assessment_and_manifest():
assert manifest["capture_target"]["platform"] == "pchome"
def _write_capture_receipt(root, *, platform="pchome", manifest_id="manifest-1"):
target_dir = root / platform / manifest_id
tiles_dir = target_dir / "tiles"
tiles_dir.mkdir(parents=True)
screenshot = target_dir / "fullpage.png"
tile_0 = tiles_dir / "tile_000.png"
tile_1 = tiles_dir / "tile_001.png"
screenshot.write_bytes(b"png")
tile_0.write_bytes(b"png")
tile_1.write_bytes(b"png")
receipt = {
"success": True,
"policy": "read_only_pixelrag_visual_capture_artifact_v1",
"status": "captured",
"generated_at": datetime.now(timezone.utc).isoformat(),
"manifest_id": manifest_id,
"capture_target": {
"platform": platform,
"crawler": "PChomeCrawler.visual_fallback",
"trigger_reason": "parser_empty",
"url": "https://24h.pchome.com.tw/",
},
"http_status": 200,
"page_metrics": {
"final_url": "https://24h.pchome.com.tw/",
"title": "PChome 24h購物",
},
"tile_plan": {
"planned_tile_count": 6,
"emitted_tile_count": 2,
},
"files": [
{"kind": "fullpage_screenshot", "path": str(screenshot)},
{"kind": "tile", "path": str(tile_0)},
{"kind": "tile", "path": str(tile_1)},
],
}
(target_dir / "capture_receipt.json").write_text(
json.dumps(receipt, ensure_ascii=False),
encoding="utf-8",
)
return target_dir / "capture_receipt.json"
def test_pixelrag_visual_evidence_readback_reports_latest_receipt(tmp_path):
from services.pixelrag_crawler_integration_service import (
READBACK_POLICY,
build_pixelrag_visual_evidence_readback,
)
receipt_path = _write_capture_receipt(tmp_path)
payload = build_pixelrag_visual_evidence_readback(
artifact_root=tmp_path,
platform="pchome",
manifest_id="manifest-1",
)
assert payload["policy"] == READBACK_POLICY
assert payload["status"] == "ok"
assert payload["summary"]["receipt_count"] == 1
assert payload["summary"]["tile_file_count"] == 2
assert payload["summary"]["missing_file_count"] == 0
assert payload["summary"]["http_status"] == 200
assert payload["latest_receipt"]["path"] == str(receipt_path)
assert payload["next_machine_action"] == "keep_pixelrag_visual_evidence_receipt"
assert payload["controlled_apply"]["db_write"] is False
def test_pixelrag_visual_evidence_readback_warns_when_missing(tmp_path):
from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback
payload = build_pixelrag_visual_evidence_readback(artifact_root=tmp_path, platform="momo")
assert payload["success"] is False
assert payload["status"] == "warning"
assert payload["summary"]["receipt_count"] == 0
assert payload["next_machine_action"] == "run_pixelrag_visual_capture_worker"
def test_pixelrag_visual_evidence_readback_route_returns_payload(tmp_path, monkeypatch):
from flask import Flask
from routes import system_public_routes as routes
from services import pixelrag_crawler_integration_service as service
_write_capture_receipt(tmp_path, platform="pchome", manifest_id="route-manifest")
monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path))
app = Flask(__name__)
with app.test_request_context(
"/api/ai-automation/pixelrag-visual-evidence-readback"
"?platform=pchome&manifest_id=route-manifest&max_age_hours=999"
):
response = routes.ai_automation_pixelrag_visual_evidence_readback_api.__wrapped__()
payload = response.get_json()
assert payload["policy"] == "read_only_pixelrag_visual_evidence_readback_v1"
assert payload["status"] == "ok"
assert payload["manifest_id"] == "route-manifest"
assert payload["summary"]["tile_file_count"] == 2
def test_pixelrag_capture_worker_dry_run_plans_artifact_outputs(tmp_path):
if not shutil.which("node"):
pytest.skip("node is required for the dry-run capture worker test")