74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""Platform modules.
|
|
|
|
Each module exposes two coroutines with a fixed shape:
|
|
|
|
async def post(cdp_client, text, media_path, options) -> dict
|
|
-> {"status": "ok" | "error", "detail": "..."} (+ "screenshot" on error)
|
|
|
|
async def check_login(cdp_client) -> bool
|
|
|
|
`post` owns its own tab: it opens one, drives the flow, closes it. The queue
|
|
guarantees only one of these runs at a time, so there is one browser and one
|
|
tab in play (proposal: "sequential per-platform").
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import traceback
|
|
|
|
from ..cdp import CDPError
|
|
|
|
SCREENSHOT_DIR = "/tmp/socialmag"
|
|
|
|
|
|
def screenshot_path(platform):
|
|
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
|
|
return os.path.join(
|
|
SCREENSHOT_DIR, f"{platform}-{time.strftime('%Y%m%d-%H%M%S')}.png"
|
|
)
|
|
|
|
|
|
async def guard(platform, tab, coro):
|
|
"""Run one platform flow, turning any failure into a labelled result plus a
|
|
screenshot of the page as it actually looked. Never masks the failure."""
|
|
try:
|
|
await coro
|
|
return {"status": "ok", "detail": f"{platform}: posted"}
|
|
except CDPError as exc:
|
|
shot = None
|
|
try:
|
|
shot = await tab.screenshot(screenshot_path(platform))
|
|
except Exception: # the page may be gone; the original error still wins
|
|
pass
|
|
result = {"status": "error", "detail": f"{platform}: {exc}"}
|
|
if shot:
|
|
result["screenshot"] = shot
|
|
return result
|
|
except Exception as exc: # noqa: BLE001 — surface it, never swallow it
|
|
return {
|
|
"status": "error",
|
|
"detail": f"{platform}: unexpected {type(exc).__name__}: {exc}",
|
|
"traceback": traceback.format_exc(),
|
|
}
|
|
|
|
|
|
def require_media(platform, media_path):
|
|
"""ig/tiktok cannot post without media; nobody gets a silent no-op."""
|
|
if not media_path:
|
|
raise CDPError(f"{platform} requires a media file; none was given")
|
|
validate_media(media_path)
|
|
|
|
|
|
def validate_media(media_path):
|
|
if not media_path:
|
|
return
|
|
if not os.path.isabs(media_path):
|
|
raise CDPError(f"media path must be absolute, got {media_path!r}")
|
|
if not os.path.exists(media_path):
|
|
raise CDPError(f"media file does not exist: {media_path}")
|
|
|
|
|
|
def require_text(platform, text):
|
|
if not text or not text.strip():
|
|
raise CDPError(f"{platform} requires non-empty text")
|