Your Name
525 lines
24 KiB
Python
525 lines
24 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import html
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
try:
|
||
import fcntl
|
||
except ImportError: # pragma: no cover - production runs on Linux
|
||
fcntl = None
|
||
|
||
import aiohttp
|
||
from aiogram import Bot
|
||
from aiogram.client.session.aiohttp import AiohttpSession
|
||
from aiogram.client.telegram import TelegramAPIServer
|
||
from PIL import Image, ImageOps
|
||
|
||
from vk_parser_app.config import settings
|
||
from vk_parser_app.db import close_pool, fetch_setting, get_pool
|
||
|
||
|
||
SEPARATOR_RE = re.compile(r"^━+$")
|
||
HASHTAG_LINE_RE = re.compile(r"^(?:#[\wа-яё]+\s*)+$", re.IGNORECASE)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Publish selected accepted posts to Ghost")
|
||
group = parser.add_mutually_exclusive_group(required=True)
|
||
group.add_argument("--ids", help="Comma-separated raw_post ids")
|
||
group.add_argument("--all-ready", action="store_true", help="Publish every photo-ready post not yet published to site")
|
||
group.add_argument("--all-ready-videos", action="store_true", help="Publish video-only ready posts")
|
||
parser.add_argument("--ghost-url", default="https://f-n8.ru")
|
||
parser.add_argument("--key-file", default="/opt/vk-parser/.site-poster-key")
|
||
parser.add_argument("--limit", type=int, default=0, help="Maximum posts to publish in this run; 0 means unlimited")
|
||
parser.add_argument("--retry-failed", action="store_true", help="Include previous site publish_failed rows")
|
||
return parser.parse_args()
|
||
|
||
|
||
def ghost_token(key: str) -> str:
|
||
key_id, secret = key.strip().split(":", 1)
|
||
now = int(time.time())
|
||
|
||
def encode(value: dict[str, Any] | bytes) -> str:
|
||
raw = value if isinstance(value, bytes) else json.dumps(value, separators=(",", ":")).encode()
|
||
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
||
|
||
header = encode({"alg": "HS256", "typ": "JWT", "kid": key_id})
|
||
payload = encode({"iat": now, "exp": now + 300, "aud": "/admin/"})
|
||
signature = encode(hmac.new(bytes.fromhex(secret), f"{header}.{payload}".encode(), hashlib.sha256).digest())
|
||
return f"{header}.{payload}.{signature}"
|
||
|
||
|
||
def split_post_text(value: str) -> tuple[str, str]:
|
||
lines = [line.strip() for line in str(value or "").replace("\r", "").split("\n")]
|
||
while lines and not lines[0]:
|
||
lines.pop(0)
|
||
title = lines.pop(0) if lines else "Новость Forma N8"
|
||
while lines and (not lines[0] or SEPARATOR_RE.fullmatch(lines[0])):
|
||
lines.pop(0)
|
||
while lines and (not lines[-1] or HASHTAG_LINE_RE.fullmatch(lines[-1])):
|
||
lines.pop()
|
||
body = "\n".join(lines).strip()
|
||
return title[:255], body
|
||
|
||
|
||
def paragraphs(value: str) -> str:
|
||
chunks = [chunk.strip() for chunk in re.split(r"\n\s*\n", value) if chunk.strip()]
|
||
return "".join(f"<p>{html.escape(chunk).replace(chr(10), '<br>')}</p>" for chunk in chunks)
|
||
|
||
|
||
def excerpt(value: str, limit: int = 260) -> str:
|
||
clean = re.sub(r"\s+", " ", value).strip()
|
||
if len(clean) <= limit:
|
||
return clean
|
||
return clean[: limit - 1].rsplit(" ", 1)[0] + "…"
|
||
|
||
|
||
def compress_photo(data: bytes) -> tuple[bytes, int, int]:
|
||
with Image.open(io.BytesIO(data)) as source:
|
||
image = ImageOps.exif_transpose(source)
|
||
if image.mode not in {"RGB", "RGBA"}:
|
||
image = image.convert("RGB")
|
||
image.thumbnail((2000, 2000), Image.Resampling.LANCZOS)
|
||
output = io.BytesIO()
|
||
image.save(output, "WEBP", quality=82, method=6)
|
||
return output.getvalue(), image.width, image.height
|
||
|
||
|
||
def video_thumbnail(data: bytes) -> tuple[bytes, int, int]:
|
||
video_path = ""
|
||
image_path = ""
|
||
try:
|
||
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as video_file:
|
||
video_file.write(data)
|
||
video_path = video_file.name
|
||
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as image_file:
|
||
image_path = image_file.name
|
||
command = [
|
||
"ffmpeg", "-y", "-ss", "1", "-i", video_path,
|
||
"-frames:v", "1", "-vf", "scale='min(2000,iw)':-2", "-q:v", "3", image_path,
|
||
]
|
||
result = subprocess.run(command, capture_output=True, timeout=90)
|
||
if result.returncode != 0 or not Path(image_path).exists() or Path(image_path).stat().st_size == 0:
|
||
command[3] = "0"
|
||
result = subprocess.run(command, capture_output=True, timeout=90)
|
||
if result.returncode != 0:
|
||
raise RuntimeError("ffmpeg thumbnail extraction failed")
|
||
return compress_photo(Path(image_path).read_bytes())
|
||
finally:
|
||
for path in (video_path, image_path):
|
||
if path:
|
||
try:
|
||
os.unlink(path)
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
def normalize_video(data: bytes, duration: int, height: int | None) -> bytes:
|
||
max_bytes = 40 * 1024 * 1024
|
||
if len(data) <= max_bytes and (not height or height <= 720):
|
||
return data
|
||
input_path = ""
|
||
output_path = ""
|
||
try:
|
||
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as input_file:
|
||
input_file.write(data)
|
||
input_path = input_file.name
|
||
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as output_file:
|
||
output_path = output_file.name
|
||
safe_duration = max(1, int(duration or 1))
|
||
target_total_kbps = int((38 * 1024 * 8) / safe_duration)
|
||
video_kbps = max(280, min(2800, target_total_kbps - 96))
|
||
command = [
|
||
"ffmpeg", "-y", "-i", input_path,
|
||
"-map_metadata", "-1", "-c:v", "libx264", "-preset", "fast",
|
||
"-b:v", f"{video_kbps}k", "-maxrate", f"{video_kbps}k", "-bufsize", f"{video_kbps * 2}k",
|
||
]
|
||
if height and int(height) > 720:
|
||
command.extend(["-vf", "scale=-2:720"])
|
||
command.extend(["-c:a", "aac", "-b:a", "96k", "-movflags", "+faststart", output_path])
|
||
result = subprocess.run(command, capture_output=True, timeout=max(180, safe_duration * 3))
|
||
if result.returncode != 0 or not Path(output_path).exists():
|
||
raise RuntimeError("ffmpeg video normalization failed")
|
||
normalized = Path(output_path).read_bytes()
|
||
if len(normalized) > 48 * 1024 * 1024:
|
||
raise RuntimeError(f"normalized video remains too large: {len(normalized)} bytes")
|
||
return normalized
|
||
finally:
|
||
for path in (input_path, output_path):
|
||
if path:
|
||
try:
|
||
os.unlink(path)
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
def gallery_html(images: list[dict[str, Any]], title: str) -> str:
|
||
if not images:
|
||
return ""
|
||
if len(images) == 1:
|
||
image = images[0]
|
||
return (
|
||
'<figure class="kg-card kg-image-card kg-width-wide">'
|
||
f'<img src="{html.escape(image["url"], quote=True)}" '
|
||
f'width="{image["width"]}" height="{image["height"]}" '
|
||
f'alt="{html.escape(title, quote=True)}" loading="lazy">'
|
||
"</figure>"
|
||
)
|
||
rows = []
|
||
for offset in range(0, len(images), 3):
|
||
cells = []
|
||
for image in images[offset : offset + 3]:
|
||
cells.append(
|
||
'<div class="kg-gallery-image">'
|
||
f'<img src="{html.escape(image["url"], quote=True)}" '
|
||
f'width="{image["width"]}" height="{image["height"]}" '
|
||
f'alt="{html.escape(title, quote=True)}" loading="lazy">'
|
||
"</div>"
|
||
)
|
||
rows.append('<div class="kg-gallery-row">' + "".join(cells) + "</div>")
|
||
return '<figure class="kg-card kg-gallery-card kg-width-wide"><div class="kg-gallery-container">' + "".join(rows) + "</div></figure>"
|
||
|
||
|
||
def videos_html(videos: list[dict[str, Any]], title: str, poster_url: str) -> str:
|
||
return "".join(
|
||
'<figure class="kg-card kg-video-card kg-width-wide">'
|
||
f'<video src="{html.escape(video["url"], quote=True)}" '
|
||
f'poster="{html.escape(poster_url, quote=True)}" controls playsinline preload="metadata" '
|
||
f'aria-label="{html.escape(title, quote=True)}"></video>'
|
||
"</figure>"
|
||
for video in videos
|
||
)
|
||
|
||
|
||
def lexical_paragraph(text: str) -> dict[str, Any]:
|
||
return {
|
||
"children": [{"detail": 0, "format": 0, "mode": "normal", "style": "", "text": text, "type": "extended-text", "version": 1}],
|
||
"direction": None,
|
||
"format": "",
|
||
"indent": 0,
|
||
"type": "paragraph",
|
||
"version": 1,
|
||
}
|
||
|
||
|
||
def video_lexical(videos: list[dict[str, Any]], body: str, thumbnail: dict[str, Any]) -> str:
|
||
children: list[dict[str, Any]] = []
|
||
for video in videos:
|
||
children.append(
|
||
{
|
||
"type": "video",
|
||
"version": 1,
|
||
"src": video["url"],
|
||
"caption": "",
|
||
"fileName": video["file_name"],
|
||
"mimeType": "video/mp4",
|
||
"width": video.get("width"),
|
||
"height": video.get("height"),
|
||
"duration": video.get("duration") or 0,
|
||
"thumbnailSrc": thumbnail["url"],
|
||
"customThumbnailSrc": "",
|
||
"thumbnailWidth": thumbnail["width"],
|
||
"thumbnailHeight": thumbnail["height"],
|
||
"cardWidth": "wide",
|
||
"loop": False,
|
||
}
|
||
)
|
||
children.extend(lexical_paragraph(chunk.strip()) for chunk in re.split(r"\n\s*\n", body) if chunk.strip())
|
||
return json.dumps(
|
||
{"root": {"children": children, "direction": None, "format": "", "indent": 0, "type": "root", "version": 1}},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
|
||
class GhostClient:
|
||
def __init__(self, session: aiohttp.ClientSession, base_url: str, key: str) -> None:
|
||
self.session = session
|
||
self.base_url = base_url.rstrip("/")
|
||
self.key = key
|
||
|
||
def headers(self) -> dict[str, str]:
|
||
return {"Authorization": f"Ghost {ghost_token(self.key)}", "Accept-Version": "v6.0"}
|
||
|
||
async def upload_image(self, data: bytes, name: str, ref: str) -> str:
|
||
form = aiohttp.FormData()
|
||
form.add_field("file", data, filename=name, content_type="image/webp")
|
||
form.add_field("purpose", "image")
|
||
form.add_field("ref", ref)
|
||
async with self.session.post(
|
||
f"{self.base_url}/ghost/api/admin/images/upload/", data=form, headers=self.headers()
|
||
) as response:
|
||
payload = await response.json(content_type=None)
|
||
if response.status >= 300:
|
||
raise RuntimeError(f"Ghost image upload {response.status}: {payload}")
|
||
return str(payload["images"][0]["url"])
|
||
|
||
async def upload_media(self, data: bytes, name: str, content_type: str) -> str:
|
||
form = aiohttp.FormData()
|
||
form.add_field("file", data, filename=name, content_type=content_type)
|
||
async with self.session.post(
|
||
f"{self.base_url}/ghost/api/admin/media/upload/", data=form, headers=self.headers()
|
||
) as response:
|
||
payload = await response.json(content_type=None)
|
||
if response.status >= 300:
|
||
raise RuntimeError(f"Ghost media upload {response.status}: {payload}")
|
||
return str(payload["media"][0]["url"])
|
||
|
||
async def create_post(self, post: dict[str, Any]) -> dict[str, Any]:
|
||
suffix = "/" if "lexical" in post else "/?source=html"
|
||
async with self.session.post(
|
||
f"{self.base_url}/ghost/api/admin/posts{suffix}",
|
||
json={"posts": [post]},
|
||
headers=self.headers(),
|
||
) as response:
|
||
payload = await response.json(content_type=None)
|
||
if response.status >= 300:
|
||
raise RuntimeError(f"Ghost post create {response.status}: {payload}")
|
||
return dict(payload["posts"][0])
|
||
|
||
|
||
async def main() -> None:
|
||
args = parse_args()
|
||
key = Path(args.key_file).read_text(encoding="utf-8").strip()
|
||
pool = await get_pool()
|
||
if args.all_ready or args.all_ready_videos:
|
||
media_condition = (
|
||
"""EXISTS (
|
||
SELECT 1 FROM raw_post_media rpm
|
||
WHERE rpm.raw_post_id=rp.id AND rpm.media_type='video'
|
||
AND COALESCE(rpm.editor_hidden,FALSE)=FALSE
|
||
AND COALESCE(rpm.tg_file_id,rpm.storage_attachment_id,'')<>''
|
||
) AND NOT EXISTS (
|
||
SELECT 1 FROM raw_post_media rpm
|
||
WHERE rpm.raw_post_id=rp.id AND rpm.media_type='photo'
|
||
AND COALESCE(rpm.editor_hidden,FALSE)=FALSE
|
||
AND COALESCE(rpm.tg_file_id,rpm.storage_attachment_id,'')<>''
|
||
)"""
|
||
if args.all_ready_videos
|
||
else """EXISTS (
|
||
SELECT 1 FROM raw_post_media rpm
|
||
WHERE rpm.raw_post_id=rp.id AND rpm.media_type='photo'
|
||
AND COALESCE(rpm.editor_hidden,FALSE)=FALSE
|
||
AND COALESCE(rpm.tg_file_id,rpm.storage_attachment_id,'')<>''
|
||
)"""
|
||
)
|
||
candidates = [
|
||
dict(row)
|
||
for row in await pool.fetch(
|
||
f"""
|
||
SELECT rp.id,
|
||
COALESCE(rp.final_category_tag,rp.rewrite_category_tag,'') AS category,
|
||
COALESCE(rp.final_source_tag,rp.rewrite_source_tag,s.tag,'') AS source
|
||
FROM raw_posts rp
|
||
JOIN sources s ON s.id=rp.source_id
|
||
JOIN content_categories cc ON cc.sort_order=COALESCE(rp.final_category_id,rp.rewrite_category_id)
|
||
LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='site'
|
||
WHERE rp.status='storage_ready'
|
||
AND rp.rewrite_status='ready'
|
||
AND COALESCE(rp.editorial_status,'review') IN ('accepted','published','publish_failed')
|
||
AND cc.site_enabled=TRUE
|
||
AND COALESCE(pp.status,'pending') {"<> 'published'" if args.retry_failed else "= 'pending'"}
|
||
AND {media_condition}
|
||
ORDER BY COALESCE(rp.reviewed_at,rp.edited_at,rp.rewritten_at,rp.created_at),rp.id
|
||
"""
|
||
)
|
||
]
|
||
recent = [
|
||
dict(row)
|
||
for row in await pool.fetch(
|
||
"""
|
||
SELECT COALESCE(rp.final_category_tag,rp.rewrite_category_tag,'') AS category,
|
||
COALESCE(rp.final_source_tag,rp.rewrite_source_tag,s.tag,'') AS source
|
||
FROM post_publications pp
|
||
JOIN raw_posts rp ON rp.id=pp.raw_post_id
|
||
JOIN sources s ON s.id=rp.source_id
|
||
WHERE pp.platform='site' AND pp.status='published'
|
||
ORDER BY pp.published_at DESC NULLS LAST,pp.id DESC LIMIT 20
|
||
"""
|
||
)
|
||
]
|
||
selected: list[dict[str, Any]] = []
|
||
while candidates:
|
||
previous = selected[-1] if selected else (recent[0] if recent else {})
|
||
recent_window = recent + selected[-20:]
|
||
best = min(
|
||
range(len(candidates)),
|
||
key=lambda index: (
|
||
int(bool(candidates[index]["category"]) and candidates[index]["category"] == previous.get("category"))
|
||
+ int(bool(candidates[index]["source"]) and candidates[index]["source"] == previous.get("source")),
|
||
sum(3 for row in recent_window if row.get("category") == candidates[index]["category"])
|
||
+ sum(4 for row in recent_window if row.get("source") == candidates[index]["source"]),
|
||
index,
|
||
),
|
||
)
|
||
selected.append(candidates.pop(best))
|
||
ids = [int(row["id"]) for row in selected]
|
||
if args.limit > 0:
|
||
ids = ids[: args.limit]
|
||
else:
|
||
ids = [int(item) for item in str(args.ids or "").split(",") if item.strip()]
|
||
if not ids:
|
||
print("NO_READY_POSTS")
|
||
await close_pool()
|
||
return
|
||
print(f"QUEUE {len(ids)}")
|
||
rows = await pool.fetch(
|
||
"""
|
||
SELECT rp.*, s.name AS source_name, s.tag AS source_tag,
|
||
cc.site_name, cc.site_slug, cc.site_enabled,
|
||
pp.status AS site_publication_status
|
||
FROM raw_posts rp
|
||
JOIN sources s ON s.id=rp.source_id
|
||
JOIN content_categories cc ON cc.sort_order=COALESCE(rp.final_category_id, rp.rewrite_category_id)
|
||
LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='site'
|
||
WHERE rp.id=ANY($1::bigint[])
|
||
""",
|
||
ids,
|
||
)
|
||
by_id = {int(row["id"]): dict(row) for row in rows}
|
||
local_api = str(await fetch_setting("local_bot_api_url", settings.local_bot_api_url) or "").strip()
|
||
bot_session = AiohttpSession(api=TelegramAPIServer.from_base(local_api.rstrip("/"), is_local=True)) if local_api else AiohttpSession()
|
||
bot = Bot(settings.tg_bot_token, session=bot_session)
|
||
timeout = aiohttp.ClientTimeout(total=120)
|
||
published_base = datetime.now(timezone.utc) - timedelta(minutes=len(ids))
|
||
try:
|
||
async with aiohttp.ClientSession(timeout=timeout) as http_session:
|
||
ghost = GhostClient(http_session, args.ghost_url, key)
|
||
for position, raw_post_id in enumerate(ids):
|
||
post = by_id.get(raw_post_id)
|
||
if not post:
|
||
print(f"SKIP {raw_post_id}: missing or unmapped category")
|
||
continue
|
||
if not post.get("site_enabled"):
|
||
print(f"SKIP {raw_post_id}: category disabled for site")
|
||
continue
|
||
if post.get("site_publication_status") == "published":
|
||
print(f"SKIP {raw_post_id}: already published")
|
||
continue
|
||
try:
|
||
media_rows = await pool.fetch(
|
||
"""
|
||
SELECT * FROM raw_post_media
|
||
WHERE raw_post_id=$1 AND media_type IN ('photo','video')
|
||
AND COALESCE(editor_hidden,FALSE)=FALSE
|
||
AND COALESCE(tg_file_id,storage_attachment_id,'')<>''
|
||
ORDER BY sort_order,id
|
||
""",
|
||
raw_post_id,
|
||
)
|
||
uploaded_photos = []
|
||
uploaded_videos = []
|
||
thumbnail = None
|
||
for index, media in enumerate(media_rows):
|
||
file_id = str(media["tg_file_id"] or media["storage_attachment_id"])
|
||
telegram_file = await bot.get_file(file_id)
|
||
buffer = io.BytesIO()
|
||
await bot.download_file(telegram_file.file_path, destination=buffer)
|
||
data = buffer.getvalue()
|
||
if media["media_type"] == "photo":
|
||
compressed, width, height = compress_photo(data)
|
||
url = await ghost.upload_image(compressed, f"post-{raw_post_id}-{index + 1}.webp", f"raw-post-{raw_post_id}-{index + 1}")
|
||
uploaded_photos.append({"type": "photo", "url": url, "width": width, "height": height, "bytes": len(compressed)})
|
||
else:
|
||
data = normalize_video(data, int(media.get("duration_sec") or 0), media.get("height"))
|
||
url = await ghost.upload_media(data, f"post-{raw_post_id}-{index + 1}.mp4", "video/mp4")
|
||
uploaded_videos.append({
|
||
"type": "video", "url": url, "bytes": len(data),
|
||
"file_name": f"post-{raw_post_id}-{index + 1}.mp4",
|
||
"width": media.get("width"), "height": media.get("height"),
|
||
"duration": media.get("duration_sec") or 0,
|
||
})
|
||
if thumbnail is None:
|
||
compressed, width, height = video_thumbnail(data)
|
||
thumb_url = await ghost.upload_image(compressed, f"post-{raw_post_id}-video-cover.webp", f"raw-post-{raw_post_id}-video-cover")
|
||
thumbnail = {"type": "thumbnail", "url": thumb_url, "width": width, "height": height, "bytes": len(compressed)}
|
||
feature = uploaded_photos[0] if uploaded_photos else thumbnail
|
||
if not feature:
|
||
raise RuntimeError("no downloadable site media")
|
||
title, body = split_post_text(post.get("final_text") or post.get("rewritten_text") or "")
|
||
producer = str(post.get("source_name") or "").strip()
|
||
content = gallery_html(uploaded_photos[1:], title) + videos_html(uploaded_videos, title, feature["url"]) + paragraphs(body)
|
||
attachments = uploaded_photos + uploaded_videos + ([thumbnail] if thumbnail else [])
|
||
payload = {
|
||
"title": title,
|
||
"status": "published",
|
||
"published_at": (published_base + timedelta(minutes=position)).isoformat(),
|
||
"custom_excerpt": excerpt(body),
|
||
"meta_description": excerpt(body),
|
||
"feature_image": feature["url"],
|
||
"feature_image_alt": title,
|
||
"tags": [
|
||
{"name": str(post["site_name"]), "slug": str(post["site_slug"])},
|
||
{"name": f"#producer:{producer}"},
|
||
{"name": f"#raw-post:{raw_post_id}"},
|
||
],
|
||
}
|
||
if uploaded_videos and not uploaded_photos:
|
||
payload["lexical"] = video_lexical(uploaded_videos, body, thumbnail)
|
||
else:
|
||
payload["html"] = content
|
||
created = await ghost.create_post(payload)
|
||
await pool.execute(
|
||
"""
|
||
INSERT INTO post_publications(raw_post_id,platform,status,target_id,external_id,url,error,attachments_json,published_at)
|
||
VALUES($1,'site','published',$2,$3,$4,NULL,$5::jsonb,NOW())
|
||
ON CONFLICT(raw_post_id,platform) DO UPDATE SET
|
||
status='published',target_id=EXCLUDED.target_id,external_id=EXCLUDED.external_id,
|
||
url=EXCLUDED.url,error=NULL,attachments_json=EXCLUDED.attachments_json,
|
||
published_at=NOW(),updated_at=NOW()
|
||
""",
|
||
raw_post_id,
|
||
args.ghost_url,
|
||
str(created["id"]),
|
||
str(created["url"]),
|
||
json.dumps(attachments, ensure_ascii=False),
|
||
)
|
||
print(f"PUBLISHED {raw_post_id} {created['url']} media={len(attachments)} bytes={sum(x['bytes'] for x in attachments)}")
|
||
except Exception as exc:
|
||
await pool.execute(
|
||
"""
|
||
INSERT INTO post_publications(raw_post_id,platform,status,target_id,error,updated_at)
|
||
VALUES($1,'site','publish_failed',$2,$3,NOW())
|
||
ON CONFLICT(raw_post_id,platform) DO UPDATE SET
|
||
status='publish_failed',target_id=EXCLUDED.target_id,error=EXCLUDED.error,updated_at=NOW()
|
||
""",
|
||
raw_post_id,
|
||
args.ghost_url,
|
||
str(exc)[:2000],
|
||
)
|
||
print(f"FAILED {raw_post_id}: {exc}")
|
||
finally:
|
||
await bot.session.close()
|
||
await close_pool()
|
||
|
||
|
||
def run_locked() -> None:
|
||
if fcntl is None:
|
||
asyncio.run(main())
|
||
return
|
||
with Path("/tmp/forma-n8-site-poster.lock").open("w") as lock_file:
|
||
try:
|
||
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
print("ALREADY_RUNNING")
|
||
return
|
||
asyncio.run(main())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_locked()
|