#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Run the Facebook Page Follow + Messenger DM workflow. This is the module-2 orchestration entrypoint: 1. Read the Facebook sheet and generate an English outreach preview. 2. Run the AdsPower/Playwright Follow + Messenger DM executor. 3. Print the full human-readable preview before any real action. Real browser actions require --confirm-send, which passes one batch-level confirmation to the executor. Workbook updates require --write-workbook. The AdsPower browser is always kept open. """ from __future__ import annotations import argparse import importlib.util import json import sys from datetime import datetime from pathlib import Path from typing import Any, List, Optional, Sequence sys.path.append(str(Path(__file__).resolve().parents[1])) from common.artifact_manager import resolve_artifact_path SCRIPT_DIR = Path(__file__).resolve().parent SKILL_ROOT = SCRIPT_DIR.parents[1] PROJECT_ROOT = Path.cwd() def load_module(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise ImportError(f"Cannot load module: {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module prepare_mod = load_module("prepare_facebook_outreach", SCRIPT_DIR / "prepare_facebook_outreach.py") send_mod = load_module("send_facebook_outreach", SCRIPT_DIR / "send_facebook_outreach.py") def timestamp() -> str: return datetime.now().strftime("%Y%m%d_%H%M%S") def resolve_output_path(raw: str, output_dir: str, run_id: str = "") -> Path: if output_dir and not raw: base = Path(output_dir).expanduser() if not base.is_absolute(): base = Path.cwd() / base return base / f"facebook_follow_dm_preview_{timestamp()}.json" return resolve_artifact_path( raw, kind="facebook_follow_dm", default_name=f"facebook_follow_dm_preview_{timestamp()}.json", run_id=run_id or None, ) def build_prepare_args(args: argparse.Namespace, preview_path: Path) -> List[str]: prepare_args: List[str] = [ "--sheet", args.sheet, "--output", str(preview_path), ] if args.excel: prepare_args.extend(["--excel", args.excel]) for item in args.filter: prepare_args.extend(["--filter", item]) if args.max_contacts > 0: prepare_args.extend(["--sample", str(args.max_contacts)]) if args.include_sent: prepare_args.append("--include-sent") if args.seed is not None: prepare_args.extend(["--seed", str(args.seed)]) return prepare_args def build_send_args(args: argparse.Namespace, preview_path: Path) -> List[str]: send_args: List[str] = [ "--preview", str(preview_path), "--profile-id", args.profile_id, "--ads-power-url", args.ads_power_url, "--action", "follow_dm", "--max-per-run", str(args.max_contacts), "--risk-profile", args.risk_profile, "--daily-follow-limit", str(args.daily_follow_limit), "--daily-dm-limit", str(args.daily_dm_limit), "--session-max", str(args.session_max), ] if args.confirm_send: send_args.extend(["--confirm", "--batch-confirmed"]) if args.use_open_page: send_args.append("--use-open-page") send_args.append("--keep-browser-open") if args.dry_run_schedule: send_args.append("--dry-run-schedule") if args.write_workbook: send_args.append("--write-workbook") else: send_args.append("--no-write-workbook") return send_args def read_preview_summary(preview_path: Path) -> dict[str, Any]: data = json.loads(preview_path.read_text(encoding="utf-8")) return data.get("summary", {}) def render_human_preview(preview_path: Path, max_chars: int = 16000) -> str: data = json.loads(preview_path.read_text(encoding="utf-8")) lines: List[str] = [] summary = data.get("summary", {}) lines.append("# Facebook Follow + Messenger DM 预览") lines.append(f"准备发送：{summary.get('ready_to_send', 0)} 条；跳过：{summary.get('skipped', 0)} 条") lines.append("") for idx, item in enumerate(data.get("items", []), start=1): fp = item.get("formatted_preview", {}) or {} lines.append(f"## {idx}. {item.get('dealer_name', '')}") lines.append(f"主页：{item.get('page_url', '')}") lines.append(f"客户判断：{fp.get('客户判断') or item.get('customer_judgment_cn', '')}") lines.append(f"推荐切入点：{fp.get('推荐切入点') or item.get('outreach_angle_cn', '')}") lines.append("英文首轮私信：") lines.append(item.get("recommended_message") or item.get("english_first_dm", "")) lines.append(f"风险提示：{fp.get('风险提示') or item.get('risk_note', '')}") lines.append("") rendered = "\n".join(lines).strip() if len(rendered) > max_chars: rendered = rendered[:max_chars] + "\n\n[预览过长，已截断；完整 JSON 见输出文件]" return rendered def build_schedule_preview(preview_path: Path, args: argparse.Namespace) -> dict[str, Any]: data = json.loads(preview_path.read_text(encoding="utf-8")) all_items = data.get("items", []) ledger = send_mod.load_ledger(args.profile_id) profile = send_mod.PACING_PROFILES[args.risk_profile] requested = max(0, args.max_contacts) allowed = send_mod.allowed_item_count("follow_dm", requested, ledger, args.daily_follow_limit, args.daily_dm_limit, args.session_max) items = all_items[:allowed] return send_mod.build_schedule_preview(items, argparse.Namespace( confirm=args.confirm_send, risk_profile=args.risk_profile, profile_id=args.profile_id, daily_follow_limit=args.daily_follow_limit, daily_dm_limit=args.daily_dm_limit, session_max=args.session_max, action="follow_dm", ), ledger, profile) def render_schedule_markdown(schedule: dict[str, Any]) -> str: sched = schedule.get("schedule", {}) used = schedule.get("today_used", {}) remaining = schedule.get("today_remaining", {}) limits = schedule.get("limits", {}) lines = [ "# Facebook 合规执行节奏预览", f"风险档位：{schedule.get('risk_profile')}；账号：{schedule.get('profile_id')}", f"本轮客户：{sched.get('customers', 0)}；预计耗时：{sched.get('estimated_minutes_min', 0)}-{sched.get('estimated_minutes_max', 0)} 分钟", f"今日已用：Follow {used.get('follow', 0)} / DM {used.get('dm', 0)} / 失败 {used.get('failures', 0)} / 风险 {used.get('risk_events', 0)}", f"今日剩余：Follow {remaining.get('follow', 0)} / DM {remaining.get('dm', 0)}；单轮上限：{limits.get('session_max', 0)}", "等待分层：大层级 major=90-200s；小层级 minor=30-90s；技术等待 technical=0.5-8s，只用于 DOM/元素检测，不计入对外行为节奏。", "停止规则：出现验证、限流、异常活动、身份确认或失败阈值达到时立即停止，不关闭浏览器。", "", "阶段等待：", ] stages = sched.get("stages", {}) or {} stage_tiers = sched.get("stage_tiers", {}) or {} for stage, wait_range in stages.items(): tier = stage_tiers.get(stage, "technical") if isinstance(wait_range, (list, tuple)) and len(wait_range) == 2: range_text = f"{wait_range[0]}-{wait_range[1]}s" else: range_text = str(wait_range) lines.append(f"- {stage}: {tier} / {range_text}") lines.append("") for idx, item in enumerate(schedule.get("customers", []), start=1): lines.append(f"{idx}. {item.get('dealer_name', '')} - {item.get('page_url', '')}") return "\n".join(lines) def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser( description="Generate Facebook outreach preview, then run Follow + Messenger DM through AdsPower/Playwright." ) parser.add_argument("--excel", default="", help="Customer outreach workbook path. If omitted, use the workbook resolver.") parser.add_argument("--sheet", default="Facebook", help="Source sheet name. Default: Facebook.") parser.add_argument("--filter", action="append", default=[], help="Filter condition field=value. Can repeat.") parser.add_argument("--profile-id", required=True, help="AdsPower profile ID already logged into Facebook.") parser.add_argument("--ads-power-url", default=send_mod.DEFAULT_ADS_POWER_URL, help="AdsPower local API URL.") parser.add_argument("--max-contacts", type=int, default=3, help="Maximum customers for this run. Default: 3.") parser.add_argument("--output", default="", help="Preview JSON output path. Default: previews/facebook_follow_dm_preview_