Update UI and migrations, fix dockerfile

This commit is contained in:
Your Name
2026-07-18 21:29:05 +05:00
parent 580c75cbf6
commit 26ef145ccf
95 changed files with 9541 additions and 111 deletions
+1027 -44
View File
File diff suppressed because it is too large Load Diff
+10
View File
@@ -11,6 +11,10 @@ POST_STATUS_STORAGE_READY = "storage_ready"
POST_STATUS_SKIPPED = "skipped"
POST_STATUS_FAILED = "failed"
PUBLICATION_STATUS_PENDING = "pending"
PUBLICATION_STATUS_PUBLISHED = "published"
PUBLICATION_STATUS_FAILED = "publish_failed"
MEDIA_STATUS_PENDING = "pending"
MEDIA_STATUS_UPLOADED = "uploaded"
MEDIA_STATUS_LINK_ONLY = "link_only"
@@ -28,3 +32,9 @@ WORKER_PARSER = "vk-parser"
WORKER_STORAGE_UPLOADER = "vk-storage-uploader"
WORKER_AI_QUALIFIER = "ai-qualifier"
WORKER_AI_WRITER = "ai-writer"
WORKER_TG_POSTER = "tg-poster"
WORKER_TG_REACTOR = "tg-reactor"
WORKER_VK_POSTER = "vk-poster"
WORKER_MAX_POSTER = "max-poster"
WORKER_SITE_POSTER = "site-poster"
WORKER_DAILY_REPORT = "daily-report"
+8 -2
View File
@@ -3,6 +3,7 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="icon" type="image/png" href="/favi.png">
<title>{{ title or "VK Parser Admin" }}</title>
<style>
:root { color-scheme: light; --bg:#f4f6f8; --panel:#fff; --panel-soft:#f8fafc; --line:#d8dee8; --line-strong:#c8d0dc; --text:#20242b; --muted:#647086; --accent:#245fd6; --accent-soft:#eef4ff; --danger:#b42318; --ok:#027a48; --shadow:0 1px 2px rgba(16,24,40,.04), 0 10px 30px rgba(16,24,40,.06); }
@@ -65,7 +66,7 @@
.source-cell a { font-weight:700; }
.stage-cell { min-width:112px; }
.raw-table { table-layout:fixed; }
.raw-table th:nth-child(1), .raw-table td:nth-child(1) { width:58px; }
.raw-table th:nth-child(1), .raw-table td:nth-child(1) { width:82px; }
.raw-table th:nth-child(2), .raw-table td:nth-child(2) { width:170px; }
.raw-table th:nth-child(3), .raw-table td:nth-child(3) { width:120px; }
.raw-table th:nth-child(5), .raw-table td:nth-child(5) { width:96px; }
@@ -191,6 +192,8 @@
.sort-head:hover { color:var(--accent); }
.sort-head.active:after { content:"↓"; font-size:11px; color:var(--accent); }
.sort-head.active.asc:after { content:"↑"; }
.raw-select-head, .raw-select-cell { display:flex; align-items:center; gap:8px; margin:0; }
.raw-select-head input, .raw-select-cell input { width:16px; height:16px; margin:0; flex:0 0 auto; }
.editor-list { display:grid; gap:12px; }
.editor-card { display:grid; grid-template-columns:112px minmax(0,1fr) 132px; gap:16px; align-items:start; background:var(--panel); border:1px solid var(--line); border-radius:8px; padding:14px; box-shadow:var(--shadow); }
.editor-media .media-strip { display:grid; grid-template-columns:repeat(2, 48px); overflow:visible; }
@@ -198,7 +201,9 @@
.editor-meta { display:flex; gap:8px; align-items:center; flex-wrap:wrap; color:var(--muted); font-size:13px; margin-bottom:8px; }
.editor-meta a { color:var(--text); font-weight:800; text-decoration:none; }
.editor-title-row { display:flex; gap:8px; align-items:center; flex-wrap:wrap; margin-bottom:8px; }
.editor-text { white-space:pre-wrap; max-height:190px; overflow:hidden; line-height:1.55; }
.publication-badges { display:flex; gap:6px; flex-wrap:wrap; margin:0 0 8px; }
.publication-badges .pill { text-decoration:none; }
.editor-text { white-space:pre-wrap; line-height:1.55; overflow-wrap:anywhere; }
.source-compare { margin-top:10px; border-top:1px solid var(--line); padding-top:8px; }
.source-compare summary { cursor:pointer; color:var(--muted); font-weight:700; }
.compare-grid { display:grid; grid-template-columns:1.3fr .7fr; gap:14px; margin-top:8px; }
@@ -230,6 +235,7 @@
<a href="/editor">Редакторская</a>
<a href="/prompt-test">Тест промптов</a>
<a href="/workers">Воркеры</a>
<a href="/logs">Логи</a>
<a href="/users">Пользователи</a>
</nav>
<form method="post" action="/logout"><button class="btn" type="submit">{{ user.login }} · выйти</button></form>
+29 -8
View File
@@ -69,13 +69,22 @@
<span>{{ p.posted_at_fmt or p.created_at_fmt }}</span>
</div>
<div class="editor-title-row">
<span class="pill {% if p.editorial_status == 'accepted' %}ok{% elif p.editorial_status == 'rejected' %}bad{% elif p.editorial_status == 'regenerating' %}warn{% endif %}">
<span class="pill {% if p.editorial_status in ['accepted', 'published'] %}ok{% elif p.editorial_status in ['rejected', 'publish_failed'] %}bad{% elif p.editorial_status == 'regenerating' %}warn{% endif %}" title="{{ p.publication_error or p.tg_publication_url or '' }}">
{{ p.editorial_status_label }}
</span>
<span class="pill">{{ p.review_category or "без категории" }}</span>
<span class="muted">AI {{ p.qualification_score or "?" }}/10</span>
</div>
<div class="editor-text">{{ p.review_text }}</div>
<div class="publication-badges">
{% for pub in p.publication_platforms %}
{% if pub.url %}
<a class="pill {{ pub.class }}" href="{{ pub.url }}" target="_blank" title="{{ pub.error or pub.status_label }}">{{ pub.label }} · {{ pub.status_label }}</a>
{% else %}
<span class="pill {{ pub.class }}" title="{{ pub.error or pub.status_label }}">{{ pub.label }} · {{ pub.status_label }}</span>
{% endif %}
{% endfor %}
</div>
<div class="editor-text">{{ p.publication_preview_text }}</div>
<details class="source-compare">
<summary>Исходник и AI-заметки</summary>
<div class="compare-grid">
@@ -154,10 +163,11 @@
<label><span>Категория</span>
<select name="final_category">
{% for category in categories %}
<option value="{{ category }}" {% if p.review_category == category %}selected{% endif %}>{{ category }}</option>
<option value="{{ category.name }}" data-tag="{{ category.tag }}" {% if p.review_category == category.name %}selected{% endif %}>{{ category.name }}</option>
{% endfor %}
{% if p.review_category and p.review_category not in categories %}
<option value="{{ p.review_category }}" selected>{{ p.review_category }} · старая категория</option>
{% set category_names = categories | map(attribute="name") | list %}
{% if p.review_category and p.review_category not in category_names %}
<option value="{{ p.review_category }}" data-tag="{{ p.review_category_tag }}" selected>{{ p.review_category }} · старая категория</option>
{% endif %}
</select>
</label>
@@ -165,6 +175,9 @@
<input name="final_source_tag" value="{{ p.review_source_tag or p.source_tag or '' }}">
</label>
</div>
<label><span>Предпросмотр публикации</span>
<div class="publication-preview" data-publication-preview>{{ p.publication_preview_text }}</div>
</label>
<div class="hashtag-preview" data-hashtag-preview>#{{ p.review_category_tag or p.review_category or "category" }} #{{ p.review_source_tag or p.source_tag or "source" }}</div>
<label><span>Заметка редактора</span>
<textarea name="editor_notes" rows="3">{{ p.editor_notes or "" }}</textarea>
@@ -253,9 +266,11 @@
button.addEventListener("click", () => document.getElementById(button.dataset.closeDialog)?.close());
});
document.querySelectorAll(".edit-form").forEach((form) => {
const textArea = form.querySelector('[name="final_text"]');
const category = form.querySelector('[name="final_category"]');
const source = form.querySelector('[name="final_source_tag"]');
const preview = form.querySelector("[data-hashtag-preview]");
const hashtagPreview = form.querySelector("[data-hashtag-preview]");
const publicationPreview = form.querySelector("[data-publication-preview]");
const normalize = (value, fallback) => {
const tag = String(value || "").trim().replace(/^#+/, "").toLowerCase()
.replace(/\s+/g, "_")
@@ -265,9 +280,15 @@
return tag || fallback;
};
const update = () => {
if (!preview) return;
preview.textContent = `#${normalize(category?.value, "category")} #${normalize(source?.value, "source")}`;
const option = category?.selectedOptions?.[0];
const tags = `#${normalize(option?.dataset?.tag || category?.value, "category")} #${normalize(source?.value, "source")}`;
if (hashtagPreview) hashtagPreview.textContent = tags;
if (publicationPreview) {
const cleanText = String(textArea?.value || "").trim();
publicationPreview.textContent = cleanText ? `${cleanText}\n\n${tags}` : tags;
}
};
textArea?.addEventListener("input", update);
category?.addEventListener("change", update);
source?.addEventListener("input", update);
update();
+105
View File
@@ -0,0 +1,105 @@
{% extends "base.html" %}
{% block body %}
<div class="page-head">
<div>
<h1>Логи</h1>
<div class="muted">Журнал действий и запросов админки. События старше 30 дней удаляются автоматически.</div>
</div>
</div>
<section class="panel">
<form class="filter-grid" method="get" action="/logs">
<label><span>Поиск</span><input name="q" value="{{ q }}" placeholder="action, entity, JSON"></label>
<label><span>Действие</span>
<select name="action">
<option value="">Все</option>
{% for item in actions %}
<option value="{{ item.action }}" {% if action == item.action %}selected{% endif %}>{{ item.action }} · {{ item.count }}</option>
{% endfor %}
</select>
</label>
<label><span>Пользователь</span>
<select name="actor_id">
<option value="0">Все</option>
{% for item in actors %}
<option value="{{ item.id }}" {% if actor_id == item.id %}selected{% endif %}>{{ item.login }} · {{ item.count }}</option>
{% endfor %}
</select>
</label>
<label><span>Сущность</span>
<select name="entity_type">
<option value="">Все</option>
{% for item in entity_types %}
<option value="{{ item.entity_type }}" {% if entity_type == item.entity_type %}selected{% endif %}>{{ item.entity_type }} · {{ item.count }}</option>
{% endfor %}
</select>
</label>
<label><span>С даты</span><input type="date" name="date_from" value="{{ date_from }}"></label>
<label><span>По дату</span><input type="date" name="date_to" value="{{ date_to }}"></label>
<button class="btn primary filter-submit" type="submit">Показать</button>
<a class="btn filter-submit" href="/logs">Сбросить</a>
</form>
</section>
<section class="panel">
<div class="list-footer">
<div class="muted">Всего: {{ pagination.total }}</div>
<form class="row per-page-form" method="get" action="/logs">
{% for item in preserved_query %}
{% if item.key != "per_page" %}
<input type="hidden" name="{{ item.key }}" value="{{ item.value }}">
{% endif %}
{% endfor %}
<label><span>На странице</span>
<select name="per_page" data-autosubmit>
{% for n in [25,50,100,200] %}
<option value="{{ n }}" {% if pagination.per_page == n %}selected{% endif %}>{{ n }}</option>
{% endfor %}
</select>
</label>
</form>
</div>
<table class="compact-table">
<thead>
<tr>
<th>Время</th>
<th>Пользователь</th>
<th>Действие</th>
<th>Сущность</th>
<th>Детали</th>
</tr>
</thead>
<tbody>
{% for item in logs %}
<tr>
<td class="mono">{{ item.created_at_fmt }}</td>
<td>{{ item.actor_label }}</td>
<td><span class="pill">{{ item.action }}</span></td>
<td>{{ item.entity_type }}{% if item.entity_id %} #{{ item.entity_id }}{% endif %}</td>
<td>
{% if item.details_pretty %}
<details class="nested-details">
<summary>JSON</summary>
<pre class="json-box">{{ item.details_pretty }}</pre>
</details>
{% else %}
<span class="muted"></span>
{% endif %}
</td>
</tr>
{% else %}
<tr><td colspan="5" class="empty-state">Записей нет.</td></tr>
{% endfor %}
</tbody>
</table>
{% include "pagination.html" %}
</section>
<script>
document.querySelectorAll("[data-autosubmit]").forEach((input) => {
input.addEventListener("change", () => input.form?.submit());
});
</script>
{% endblock %}
+45 -7
View File
@@ -8,12 +8,24 @@
</div>
<div class="catalog-layout">
<section class="panel catalog-main">
<div class="list-toolbar">
<div class="muted">Всего: {{ pagination.total }}</div>
</div>
<table class="raw-table">
<form id="raw-bulk-form" method="post" action="/raw/send-to-rewrite">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<input type="hidden" name="next_url" value="{{ request.url.path }}{% if request.url.query %}?{{ request.url.query }}{% endif %}">
<div class="list-toolbar">
<div>
<div class="muted">Всего: {{ pagination.total }}</div>
<div class="muted">Выбранные storage-ready посты можно вручную отправить в рерайт.</div>
</div>
<button class="btn primary" type="submit" data-bulk-submit disabled>Отправить в рерайт</button>
</div>
<table class="raw-table">
<tr>
<th><a class="sort-head {% if sort_headers.id.active %}active {{ sort_headers.id.direction }}{% endif %}" href="{{ sort_headers.id.url }}">#</a></th>
<th>
<label class="raw-select-head">
<input type="checkbox" data-raw-select-all aria-label="Выбрать все посты на странице">
<a class="sort-head {% if sort_headers.id.active %}active {{ sort_headers.id.direction }}{% endif %}" href="{{ sort_headers.id.url }}">#</a>
</label>
</th>
<th><a class="sort-head {% if sort_headers.source.active %}active {{ sort_headers.source.direction }}{% endif %}" href="{{ sort_headers.source.url }}">Источник</a></th>
<th><a class="sort-head {% if sort_headers.stage.active %}active {{ sort_headers.stage.direction }}{% endif %}" href="{{ sort_headers.stage.url }}">Этап</a></th>
<th><a class="sort-head {% if sort_headers.post.active %}active {{ sort_headers.post.direction }}{% endif %}" href="{{ sort_headers.post.url }}">Пост</a></th>
@@ -21,7 +33,12 @@
</tr>
{% for p in posts %}
<tr>
<td><a class="raw-id" href="/raw/{{ p.id }}">{{ p.id }}</a></td>
<td>
<label class="raw-select-cell">
<input type="checkbox" name="raw_post_id" value="{{ p.id }}" data-raw-select aria-label="Выбрать пост {{ p.id }}">
<a class="raw-id" href="/raw/{{ p.id }}">{{ p.id }}</a>
</label>
</td>
<td class="source-cell">
<a href="{{ p.original_url }}" target="_blank">{{ p.source_name }}</a>
<div class="muted">{{ p.source_platform }} · #{{ p.source_tag or "—" }}</div>
@@ -66,7 +83,8 @@
</td>
</tr>
{% endfor %}
</table>
</table>
</form>
<div class="list-footer">
<form class="row per-page-form" method="get" action="/raw">
{% for item in preserved_query %}
@@ -206,6 +224,26 @@
});
});
});
const rawChecks = Array.from(document.querySelectorAll("[data-raw-select]"));
const rawSelectAll = document.querySelector("[data-raw-select-all]");
const bulkSubmit = document.querySelector("[data-bulk-submit]");
function syncBulkState() {
const selected = rawChecks.filter((item) => item.checked).length;
if (bulkSubmit) {
bulkSubmit.disabled = selected === 0;
bulkSubmit.textContent = selected ? `Отправить в рерайт (${selected})` : "Отправить в рерайт";
}
if (rawSelectAll) {
rawSelectAll.checked = selected > 0 && selected === rawChecks.length;
rawSelectAll.indeterminate = selected > 0 && selected < rawChecks.length;
}
}
rawChecks.forEach((item) => item.addEventListener("change", syncBulkState));
rawSelectAll?.addEventListener("change", () => {
rawChecks.forEach((item) => { item.checked = rawSelectAll.checked; });
syncBulkState();
});
syncBulkState();
})();
</script>
{% endblock %}
+62
View File
@@ -0,0 +1,62 @@
<!doctype html>
<html lang="ru" data-theme="light">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="icon" type="image/png" href="/favi.png">
<title>{{ title or "VK Parser Admin V2" }}</title>
<!-- Tailwind CSS & DaisyUI -->
<link href="https://cdn.jsdelivr.net/npm/daisyui@4.10.1/dist/full.min.css" rel="stylesheet" type="text/css" />
<script src="https://cdn.tailwindcss.com"></script>
<!-- HTMX -->
<script src="https://unpkg.com/htmx.org@1.9.11"></script>
<!-- AlpineJS (optional but useful for local state) -->
<script defer src="https://cdn.jsdelivr.net/npm/alpinejs@3.13.8/dist/cdn.min.js"></script>
<style>
/* Custom tweaks to match Tailwind typography nicely */
body { font-family: 'Inter', system-ui, sans-serif; }
.htmx-indicator { display:none; }
.htmx-request .htmx-indicator { display:inline-block; }
.htmx-request.htmx-indicator { display:inline-block; }
</style>
</head>
<body class="bg-base-200 min-h-screen">
<div class="navbar bg-base-100 shadow-sm sticky top-0 z-50">
<div class="flex-1 px-2 mx-2">
<span class="text-lg font-bold">Parser V2</span>
<div class="hidden md:flex ml-6 gap-2">
<a href="/v2/raw" class="btn btn-ghost btn-sm {{ 'btn-active' if 'raw' in request.url.path }}">Raw-посты</a>
<!-- Temporary links to V1 for others -->
<a href="/editor" class="btn btn-ghost btn-sm">Редактор (V1)</a>
<a href="/workers" class="btn btn-ghost btn-sm">Воркеры (V1)</a>
</div>
</div>
<div class="flex-none hidden lg:block">
<ul class="menu menu-horizontal gap-2">
<li><a href="/" class="btn btn-outline btn-sm">Вернуться в V1</a></li>
<li>
<form method="post" action="/logout" class="m-0">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<button class="btn btn-ghost btn-sm text-error">Выйти</button>
</form>
</li>
</ul>
</div>
</div>
<main class="container mx-auto p-4 md:p-6 max-w-7xl">
{% block body %}{% endblock %}
</main>
<script>
// Handle HTMX errors
document.body.addEventListener('htmx:responseError', function(evt) {
alert("Ошибка при загрузке: " + evt.detail.xhr.status);
});
</script>
</body>
</html>
@@ -0,0 +1,14 @@
{% extends "v2/base.html" %}
{% block body %}
<div class="mb-6 flex justify-between items-end">
<div>
<h1 class="text-3xl font-bold mb-1">Raw-посты</h1>
<div class="text-base-content/60">Исходный пост, этап обработки и AI-оценка.</div>
</div>
</div>
<div id="content-area">
{% include "v2/raw_posts_content.html" %}
</div>
{% endblock %}
@@ -0,0 +1,152 @@
<div class="flex flex-col lg:flex-row gap-6">
<!-- Sidebar Filters -->
<aside class="w-full lg:w-72 flex-shrink-0">
<div class="card bg-base-100 shadow-sm border border-base-200">
<div class="card-body p-4">
<form hx-get="/v2/raw" hx-target="#content-area" hx-push-url="true" hx-trigger="submit, change delay:300ms">
<input type="hidden" name="sort" value="{{ sort }}">
<div class="flex justify-between items-center mb-4">
<h2 class="card-title text-lg">Фильтры</h2>
<a href="/v2/raw" class="text-sm link link-hover text-base-content/60" hx-boost="true" hx-target="#content-area">Сбросить</a>
</div>
<div class="form-control w-full mb-4">
<label class="label"><span class="label-text font-bold">Оценка AI</span></label>
<div class="flex gap-2">
<input type="number" name="score_min" placeholder="От" value="{{ score_min }}" class="input input-bordered input-sm w-full" />
<input type="number" name="score_max" placeholder="До" value="{{ score_max }}" class="input input-bordered input-sm w-full" />
</div>
</div>
<div class="form-control w-full mb-4">
<label class="label"><span class="label-text font-bold">Дата поста</span></label>
<div class="flex flex-col gap-2">
<input type="date" name="date_from" value="{{ date_from }}" class="input input-bordered input-sm w-full" />
<input type="date" name="date_to" value="{{ date_to }}" class="input input-bordered input-sm w-full" />
</div>
</div>
<div class="form-control w-full mb-4">
<label class="label"><span class="label-text font-bold">Квалификация</span></label>
<div class="bg-base-200 rounded-lg p-2 max-h-48 overflow-y-auto">
{% for item in facets.qualification_statuses %}
<label class="label cursor-pointer justify-start gap-2 p-1">
<input type="checkbox" name="qualification_status" value="{{ item.value }}" class="checkbox checkbox-sm checkbox-primary" {% if item.value in qualification_statuses %}checked{% endif %} />
<span class="label-text flex-1">{{ item.value }}</span>
<span class="badge badge-sm badge-ghost">{{ item.count }}</span>
</label>
{% endendfor %}
</div>
</div>
<div class="form-control w-full mb-4">
<label class="label"><span class="label-text font-bold">Рерайт</span></label>
<div class="bg-base-200 rounded-lg p-2 max-h-48 overflow-y-auto">
{% for item in facets.rewrite_statuses %}
<label class="label cursor-pointer justify-start gap-2 p-1">
<input type="checkbox" name="rewrite_status" value="{{ item.value }}" class="checkbox checkbox-sm checkbox-primary" {% if item.value in rewrite_statuses %}checked{% endif %} />
<span class="label-text flex-1">{{ item.value }}</span>
<span class="badge badge-sm badge-ghost">{{ item.count }}</span>
</label>
{% endfor %}
</div>
</div>
<button class="btn btn-primary w-full mt-4" type="submit">Показать</button>
</form>
</div>
</div>
</aside>
<!-- Main Content (Table) -->
<div class="flex-1 min-w-0">
<div class="card bg-base-100 shadow-sm border border-base-200">
<div class="card-body p-0">
<!-- Toolbar -->
<div class="p-4 border-b border-base-200 flex justify-between items-center bg-base-100/50">
<div class="text-sm text-base-content/70">
Всего: <span class="font-bold">{{ pagination.total }}</span>
</div>
<!-- Pagination top (optional, or per page select) -->
<div class="flex items-center gap-2">
<span class="text-sm">На странице:</span>
<select name="per_page" class="select select-bordered select-sm" form="pagination-form" hx-get="/v2/raw" hx-target="#content-area" hx-include="[name='date_from'], [name='date_to']">
{% for n in [25,50,100,200] %}
<option value="{{ n }}" {% if pagination.per_page == n %}selected{% endif %}>{{ n }}</option>
{% endfor %}
</select>
</div>
</div>
<div class="overflow-x-auto">
<table class="table table-zebra w-full text-sm">
<thead>
<tr class="bg-base-200 text-base-content">
<th class="w-16">#</th>
<th class="w-48">Источник</th>
<th class="w-32">Этап</th>
<th>Пост</th>
<th class="w-32 text-center">AI</th>
</tr>
</thead>
<tbody>
{% for p in posts %}
<tr class="hover">
<td class="font-bold">
<a href="/raw/{{ p.id }}" class="link link-primary">{{ p.id }}</a>
</td>
<td>
<a href="{{ p.original_url }}" target="_blank" class="font-medium hover:underline">{{ p.source_name }}</a>
<div class="text-xs text-base-content/60 mt-1">{{ p.source_platform }} · #{{ p.source_tag or "—" }}</div>
{% if p.posted_at_fmt %}<div class="text-xs text-base-content/60 mt-1">{{ p.posted_at_fmt }}</div>{% endif %}
</td>
<td>
<div class="badge badge-sm {% if p.stage.class == 'ok' %}badge-success{% elif p.stage.class == 'bad' %}badge-error{% elif p.stage.class == 'warn' %}badge-warning{% else %}badge-ghost{% endif %} gap-1">
{{ p.stage.label }}
</div>
{% if p.error_reason %}<div class="text-xs text-error mt-1">{{ p.error_reason }}</div>{% endif %}
</td>
<td class="max-w-md">
<div class="prose prose-sm leading-snug">
{{ p.raw_text[:200] }}{% if p.raw_text|length > 200 %}...{% endif %}
</div>
</td>
<td class="text-center">
<div class="flex flex-col items-center gap-1">
<span class="text-lg font-bold {% if p.ai_badge.class == 'ok' %}text-success{% elif p.ai_badge.class == 'bad' %}text-error{% elif p.ai_badge.class == 'warn' %}text-warning{% endif %}">
{{ p.ai_badge.label }}
</span>
<span class="text-xs text-base-content/60">{{ p.ai_badge.sublabel }}</span>
</div>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<!-- Pagination Bottom -->
<div class="p-4 border-t border-base-200 flex justify-end">
<div class="join">
{% if pagination.has_prev %}
<button class="join-item btn btn-sm" hx-get="{{ pagination.prev_url.replace('/raw', '/v2/raw') }}" hx-target="#content-area" hx-push-url="true">« Пред</button>
{% else %}
<button class="join-item btn btn-sm btn-disabled">« Пред</button>
{% endif %}
<button class="join-item btn btn-sm btn-active">Стр. {{ pagination.page }} из {{ pagination.total_pages }}</button>
{% if pagination.has_next %}
<button class="join-item btn btn-sm" hx-get="{{ pagination.next_url.replace('/raw', '/v2/raw') }}" hx-target="#content-area" hx-push-url="true">След »</button>
{% else %}
<button class="join-item btn btn-sm btn-disabled">След »</button>
{% endif %}
</div>
</div>
</div>
</div>
</div>
</div>
@@ -1,17 +1,37 @@
{% extends "base.html" %}
{% block body %}
<div class="panel" style="max-width:760px;margin:8vh auto 0;">
<h1>VK OAuth callback</h1>
{% if code %}
<p class="muted">Скопируйте code и code_verifier для обмена на access token.</p>
<h1>VK OAuth</h1>
{% if result %}
{% if result.ok %}
<p class="pill ok">Токен VK-постера сохранён</p>
<dl class="kv compact-kv">
<dt>Сообщество</dt><dd>{{ result.owner_id }}</dd>
<dt>User ID</dt><dd>{{ result.user_id or "—" }}</dd>
<dt>Живёт, сек</dt><dd>{{ result.expires_in or "—" }}</dd>
<dt>Истекает</dt><dd>{{ result.expires_at or "—" }}</dd>
<dt>Refresh token</dt><dd>{% if result.has_refresh_token %}OK{% else %}Не получен{% endif %}</dd>
<dt>wall.get</dt><dd>{% if result.wall_get_ok %}OK{% else %}Ошибка: {{ result.wall_get_error }}{% endif %}</dd>
<dt>Фото upload</dt>
<dd>
{% if result.photo_upload_server_ok %}
OK
{% else %}
Недоступно: {{ result.photo_upload_server_error }}
<div class="muted">VK выдаёт право photos для upload API отдельно; постер всё равно опубликует текст и существующие VK-вложения.</div>
{% endif %}
</dd>
</dl>
<p><a class="btn primary" href="/workers">Вернуться к воркерам</a></p>
{% else %}
<p class="pill bad">VK OAuth не завершился</p>
<p>{{ result.error_description or result.error or "Неизвестная ошибка" }}</p>
<p><a class="btn" href="/workers">Вернуться к воркерам</a> <a class="btn primary" href="/vk-oauth/start">Повторить</a></p>
{% endif %}
{% elif code %}
<p class="muted">VK вернул code. Новый обработчик должен обменивать его автоматически через <span class="mono">/vk-oauth/callback</span>.</p>
<label>code</label>
<textarea readonly style="min-height:110px;" onclick="this.select()">{{ code }}</textarea>
<label>code_verifier</label>
<textarea readonly style="min-height:110px;" onclick="this.select()">{{ code_verifier }}</textarea>
{% if device_id %}
<label>device_id</label>
<textarea readonly style="min-height:80px;" onclick="this.select()">{{ device_id }}</textarea>
{% endif %}
{% if state and expected_state and state != expected_state %}
<p class="pill bad">state не совпал, такой code лучше не использовать.</p>
{% endif %}
@@ -19,34 +39,7 @@
<p class="pill bad">{{ error }}</p>
<p>{{ error_description }}</p>
{% else %}
<p class="muted">VK вернул callback без code.</p>
<p class="muted">VK вернул callback без данных.</p>
{% endif %}
<div id="fragment-token" style="display:none;margin-top:24px;">
<h2>Access token сообщества</h2>
<p class="muted">VK передал токен в fragment URL. Скопируйте значение ниже.</p>
<textarea id="fragment-token-value" readonly style="min-height:140px;" onclick="this.select()"></textarea>
</div>
<div id="fragment-data" style="display:none;margin-top:16px;">
<h2>Параметры fragment</h2>
<textarea id="fragment-data-value" readonly style="min-height:120px;" onclick="this.select()"></textarea>
</div>
</div>
<script>
const hash = window.location.hash.replace(/^#/, "");
if (hash) {
const params = new URLSearchParams(hash);
const pairs = Array.from(params.entries());
const tokenPair = pairs.find(([key]) => key.startsWith("access_token"));
if (tokenPair) {
document.getElementById("fragment-token").style.display = "block";
document.getElementById("fragment-token-value").value = tokenPair[1];
}
if (pairs.length) {
document.getElementById("fragment-data").style.display = "block";
document.getElementById("fragment-data-value").value = pairs
.map(([key, value]) => `${key}=${value}`)
.join("\n");
}
}
</script>
{% endblock %}
+79 -3
View File
@@ -19,15 +19,18 @@
<details class="panel" open>
<summary><strong>Категории публикаций</strong></summary>
<p class="muted">Активные категории попадают в редакторскую и передаются AI-райтеру как ID, название и тэг. ID стабилен: модель возвращает его в JSON, а тэг подставляется из БД.</p>
<p class="muted">Название описывает категорию для AI, тэг используется в соцсетях. Название и URL сайта хранятся отдельно и не меняют рабочие хэштеги.</p>
<form class="category-create" method="post" action="/categories/create">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<label><span>Название</span><input name="name" placeholder="например, разгрузка" required></label>
<label><span>Тэг</span><input name="tag" placeholder="если пусто, соберётся из названия"></label>
<label><span>На сайте</span><input name="site_name" placeholder="Снаряжение" required></label>
<label><span>URL сайта</span><input name="site_slug" placeholder="snaryazhenie" pattern="[a-z0-9-]+" required></label>
<label class="checkline"><input type="checkbox" name="site_enabled" checked> Публиковать</label>
<button class="btn primary" type="submit">Добавить</button>
</form>
<table class="compact-table category-table">
<tr><th>Название</th><th>Тэг</th><th>ID</th><th>Статус</th><th></th></tr>
<tr><th>Название AI</th><th>Тэг</th><th>Название сайта</th><th>URL сайта</th><th>ID</th><th>Статус</th><th></th></tr>
{% for c in categories %}
<tr>
<td>
@@ -37,8 +40,10 @@
</form>
</td>
<td><input form="category-update-{{ c.id }}" name="tag" value="{{ c.tag }}"></td>
<td><input form="category-update-{{ c.id }}" name="site_name" value="{{ c.site_name }}" required></td>
<td><input form="category-update-{{ c.id }}" name="site_slug" value="{{ c.site_slug }}" pattern="[a-z0-9-]+" required></td>
<td><input value="{{ c.sort_order }}" type="number" readonly title="Стабильный ID категории для AI-райтера"></td>
<td>{% if c.is_active %}<span class="pill ok">активна</span>{% else %}<span class="pill">выключена</span>{% endif %}</td>
<td>{% if c.is_active %}<span class="pill ok">активна</span>{% else %}<span class="pill">выключена</span>{% endif %}<label class="checkline"><input form="category-update-{{ c.id }}" type="checkbox" name="site_enabled" {% if c.site_enabled %}checked{% endif %}> сайт</label></td>
<td class="icon-actions">
<button class="icon-btn" form="category-update-{{ c.id }}" type="submit" title="Сохранить"></button>
<form method="post" action="/categories/{{ c.id }}/toggle">
@@ -55,6 +60,75 @@
</table>
</details>
<details class="panel" open>
<summary><strong>Расписание TG-постера</strong></summary>
<p class="muted">Время публикации задаётся по Екатеринбургу. Для запуска нужны оба флага: worker <span class="mono">tg-poster</span> и настройка <span class="mono">tg_poster_enabled</span>.</p>
<form class="category-create" method="post" action="/tg-poster-schedule/create">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<label><span>Время Екб</span><input type="time" name="time" value="10:00" required></label>
<label><span>Постов</span><input type="number" name="count" value="1" min="1" max="20" required></label>
<label class="checkline"><input type="checkbox" name="enabled" checked> Активно</label>
<button class="btn primary" type="submit">Добавить</button>
</form>
<table class="compact-table category-table">
<tr><th>Время Екб</th><th>Постов</th><th>Статус</th><th></th></tr>
{% for row in tg_schedule %}
<tr>
<td>
<form id="tg-schedule-update-{{ row.id }}" method="post" action="/tg-poster-schedule/{{ row.id }}/update">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<input type="time" name="time" value="{{ row.time }}" required>
</form>
</td>
<td><input form="tg-schedule-update-{{ row.id }}" type="number" name="count" value="{{ row.count }}" min="1" max="20" required></td>
<td><label class="checkline"><input form="tg-schedule-update-{{ row.id }}" type="checkbox" name="enabled" {% if row.enabled %}checked{% endif %}> Активно</label></td>
<td class="icon-actions">
<button class="icon-btn" form="tg-schedule-update-{{ row.id }}" type="submit" title="Сохранить"></button>
<form method="post" action="/tg-poster-schedule/{{ row.id }}/delete">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<button class="icon-btn danger" type="submit" title="Удалить">×</button>
</form>
</td>
</tr>
{% endfor %}
</table>
</details>
<details class="panel" open>
<summary><strong>Расписание VK-постера</strong></summary>
<p class="muted">Время публикации задаётся по Екатеринбургу. Для запуска нужны оба флага: worker <span class="mono">vk-poster</span> и настройка <span class="mono">vk_poster_enabled</span>.</p>
<p><a class="btn" href="/vk-oauth/start">Получить VK user token</a></p>
<form class="category-create" method="post" action="/vk-poster-schedule/create">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<label><span>Время Екб</span><input type="time" name="time" value="10:05" required></label>
<label><span>Постов</span><input type="number" name="count" value="1" min="1" max="20" required></label>
<label class="checkline"><input type="checkbox" name="enabled" checked> Активно</label>
<button class="btn primary" type="submit">Добавить</button>
</form>
<table class="compact-table category-table">
<tr><th>Время Екб</th><th>Постов</th><th>Статус</th><th></th></tr>
{% for row in vk_schedule %}
<tr>
<td>
<form id="vk-schedule-update-{{ row.id }}" method="post" action="/vk-poster-schedule/{{ row.id }}/update">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<input type="time" name="time" value="{{ row.time }}" required>
</form>
</td>
<td><input form="vk-schedule-update-{{ row.id }}" type="number" name="count" value="{{ row.count }}" min="1" max="20" required></td>
<td><label class="checkline"><input form="vk-schedule-update-{{ row.id }}" type="checkbox" name="enabled" {% if row.enabled %}checked{% endif %}> Активно</label></td>
<td class="icon-actions">
<button class="icon-btn" form="vk-schedule-update-{{ row.id }}" type="submit" title="Сохранить"></button>
<form method="post" action="/vk-poster-schedule/{{ row.id }}/delete">
<input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
<button class="icon-btn danger" type="submit" title="Удалить">×</button>
</form>
</td>
</tr>
{% endfor %}
</table>
</details>
<h2>Настройки</h2>
{% for category, rows in settings|groupby("category") %}
<details class="panel" {% if category == "AI Qualifier" %}open{% endif %}>
@@ -126,6 +200,8 @@
{% endif %}
<textarea name="value" rows="12" style="width:720px; max-width:100%;">{{ s.value_json }}</textarea>
</div>
{% elif s.value_type == "json" %}
<textarea name="value" rows="5" style="width:520px; max-width:100%;" spellcheck="false">{{ s.value_json | tojson }}</textarea>
{% else %}
<input name="value" value="{{ s.value_json }}" style="width:180px;">
{% endif %}
+28
View File
@@ -11,6 +11,34 @@ def normalize_hash_tag(value: str, fallback: str = "source") -> str:
return tag or fallback
def strip_trailing_hashtag_line(text: str) -> str:
lines = str(text or "").rstrip().splitlines()
while lines and not lines[-1].strip():
lines.pop()
if not lines:
return ""
parts = lines[-1].split()
if parts and all(part.startswith("#") and normalize_hash_tag(part, "") for part in parts):
lines.pop()
return "\n".join(lines).rstrip()
def publication_hashtags(category_tag: str, source_tag: str) -> str:
tags = [
normalize_hash_tag(category_tag, "category"),
normalize_hash_tag(source_tag, "source"),
]
return " ".join(f"#{tag}" for tag in tags if tag)
def build_publication_text(text: str, category_tag: str, source_tag: str) -> str:
base = strip_trailing_hashtag_line(text)
hashtags = publication_hashtags(category_tag, source_tag)
if not hashtags:
return base
return f"{base}\n\n{hashtags}" if base else hashtags
def parse_categories(value: object) -> list[str]:
if isinstance(value, list):
raw_items = [str(item) for item in value]
+14
View File
@@ -155,6 +155,20 @@ class VKAPIClient:
raise VKAPIError(None, f"photos.saveWallPhoto returned invalid response: {json.dumps(saved)[:300]}")
return saved[0]
async def upload_wall_photo_url(self, group_id: int, url: str) -> dict:
if not self.session:
raise RuntimeError("VKAPIClient is not initialized")
async with self.session.get(url) as resp:
resp.raise_for_status()
content_type = resp.headers.get("content-type") or "image/jpeg"
data = await resp.read()
suffix = ".jpg"
if "png" in content_type:
suffix = ".png"
elif "webp" in content_type:
suffix = ".webp"
return await self.upload_wall_photo_bytes(group_id, data, filename=f"photo{suffix}")
async def create_wall_post(
self,
owner_id: int,
+32
View File
@@ -151,6 +151,22 @@ class AIQualifierWorker:
FROM raw_posts
WHERE status='storage_ready'
AND COALESCE(qualification_status, 'pending') IN ('pending', 'failed')
AND EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
)
"""
)
if int(count or 0) < batch_size:
@@ -163,6 +179,22 @@ class AIQualifierWorker:
FROM raw_posts
WHERE status='storage_ready'
AND COALESCE(qualification_status, 'pending') IN ('pending', 'failed')
AND EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
)
ORDER BY created_at ASC, id ASC
FOR UPDATE SKIP LOCKED
LIMIT $1
+71 -5
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import hashlib
import json
import re
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any
@@ -15,7 +16,14 @@ from ..constants import WORKER_AI_WRITER
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import normalize_hash_tag, parse_categories
from ..text_utils import build_publication_text, normalize_hash_tag, parse_categories
NON_TARGET_CATEGORY_ID = 18
NON_TARGET_CATEGORY_TAGS = {"нцк", "nck"}
EMOJI_LEADING_MARKER_RE = re.compile(
r"^[ \t]*(?:🧵|🎯|🛡|🧤|📦|💵|⚙\ufe0f?|🎨|🎒|🔧|👕|💪)\s+"
)
def now_utc() -> datetime:
@@ -30,6 +38,14 @@ def normalize_prompt(prompt: str) -> str:
return (prompt or "").replace("\\r\\n", "\n").replace("\\n", "\n").strip()
def strip_leading_emoji_markers(text: str) -> str:
lines = str(text or "").splitlines()
cleaned: list[str] = []
for line in lines:
cleaned.append(EMOJI_LEADING_MARKER_RE.sub("", line))
return "\n".join(cleaned).strip()
DEFAULT_CONTRACT_PROMPT = """
OUTPUT SCHEMA (return array matching input order):
{"rewrites":[{"id":123,"category_id":2,"text":"готовый текст без хэштегов","notes":"короткая заметка для редактора или null"}]}
@@ -38,7 +54,9 @@ Rules:
- Input is a JSON object with key "posts" containing accepted posts.
- Input has key "categories" with objects: id, name, tag.
- Each post includes producer_name and producer_tag. producer_name can be a manufacturer, shop, or publishing source.
- The "text" field for every rewrite must explicitly mention producer_name at least once. Do not satisfy this only through producer_tag, hashtags, notes, or metadata.
- Pick exactly one category from the categories list in input and return its numeric id as category_id.
- If the post is non-target content, choose category_id 18 ("Не целевой контент", tag "НЦК"); the app will reject it automatically.
- Do not add links or hashtags to rewritten text.
- Return JSON only. No markdown. No text outside JSON.
- Never return an empty object. Never omit the "rewrites" key.
@@ -127,7 +145,7 @@ def validate_rewrites(
post_id = int(item.get("id"))
if post_id not in expected_ids:
raise ValueError(f"AI returned unexpected id={post_id}")
text = str(item.get("text") or "").strip()
text = strip_leading_emoji_markers(str(item.get("text") or "").strip())
if len(text) < 40:
raise ValueError(f"AI returned too short rewrite for id={post_id}")
by_id = {int(c["id"]): c for c in categories if c.get("id") is not None}
@@ -144,15 +162,18 @@ def validate_rewrites(
raise ValueError(f"AI returned unknown category={category_name!r} for id={post_id}")
notes = str(item.get("notes") or "").strip()[:1000]
seen.add(post_id)
category_tag = normalize_hash_tag(category_row.get("tag") or category_row.get("name") or "", "category")
reject_by_category = int(category_row["id"]) == NON_TARGET_CATEGORY_ID or category_tag.lower() in NON_TARGET_CATEGORY_TAGS
out.append(
{
"id": post_id,
"category_id": int(category_row["id"]),
"category": str(category_row["name"]),
"category_tag": normalize_hash_tag(category_row.get("tag") or category_row.get("name") or "", "category"),
"category_tag": category_tag,
"source_tag": normalize_hash_tag(source_tags_by_id.get(post_id) or "", "source"),
"text": text,
"notes": notes,
"reject_by_category": reject_by_category,
}
)
missing = expected_ids - seen
@@ -209,6 +230,22 @@ class AIWriterWorker:
WHERE status='storage_ready'
AND qualification_status='accepted'
AND COALESCE(rewrite_status, 'pending') IN ('pending', 'failed')
AND EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
)
"""
)
if int(count or 0) < batch_size:
@@ -222,6 +259,22 @@ class AIWriterWorker:
WHERE status='storage_ready'
AND qualification_status='accepted'
AND COALESCE(rewrite_status, 'pending') IN ('pending', 'failed')
AND EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1
FROM raw_post_media m
WHERE m.raw_post_id=raw_posts.id
AND COALESCE(m.editor_hidden, FALSE)=FALSE
AND m.media_type IN ('photo', 'video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
)
ORDER BY qualified_at ASC NULLS LAST, id ASC
FOR UPDATE SKIP LOCKED
LIMIT $1
@@ -316,6 +369,9 @@ class AIWriterWorker:
async def apply_rewrites(self, batch_id: int, rewrites: list[dict], model: str, prompt: str) -> None:
for item in rewrites:
final_text = build_publication_text(item["text"], item["category_tag"], item["source_tag"])
editorial_status = "rejected" if item.get("reject_by_category") else "review"
editor_notes = "Отклонено AI: Не целевой контент" if item.get("reject_by_category") else None
await self.pool.execute(
"""
UPDATE raw_posts
@@ -329,12 +385,19 @@ class AIWriterWorker:
rewrite_category_tag=$8,
rewrite_source_tag=$9,
rewrite_category_id=$10,
editorial_status='review',
final_text=$2,
editorial_status=$11,
final_text=$12,
final_category=$7,
final_category_tag=$8,
final_source_tag=$9,
final_category_id=$10,
editor_notes=$13,
qualification_status=CASE WHEN $11='rejected' THEN 'rejected' ELSE qualification_status END,
qualification_decision=CASE WHEN $11='rejected' THEN 'rejected' ELSE qualification_decision END,
qualification_model_decision=CASE WHEN $11='rejected' THEN 'rejected' ELSE qualification_model_decision END,
qualification_reason=CASE WHEN $11='rejected' THEN 'AI writer selected non-target category' ELSE qualification_reason END,
qualification_reject_tag=CASE WHEN $11='rejected' THEN 'non_target_content' ELSE qualification_reject_tag END,
reviewed_at=CASE WHEN $11='rejected' THEN NOW() ELSE reviewed_at END,
rewritten_at=NOW(),
updated_at=NOW()
WHERE id=$1
@@ -349,6 +412,9 @@ class AIWriterWorker:
item["category_tag"],
item["source_tag"],
item["category_id"],
editorial_status,
final_text,
editor_notes,
)
async def call_ai(
+457
View File
@@ -0,0 +1,457 @@
from __future__ import annotations
import argparse
import asyncio
import json
import os
import re
import sys
from datetime import date, datetime, time, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo
from aiogram import Bot
from aiogram.exceptions import TelegramRetryAfter
from loguru import logger
from ..config import settings
from ..constants import WORKER_DAILY_REPORT
from ..db import fetch_bool_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MESSAGE_LIMIT = 4096
def valid_time(value: str) -> bool:
return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))
def parse_recipients(value: Any) -> list[int]:
raw_items: list[Any]
if isinstance(value, list):
raw_items = value
elif isinstance(value, str):
raw = value.strip()
if not raw:
return []
try:
parsed = json.loads(raw)
raw_items = parsed if isinstance(parsed, list) else [parsed]
except json.JSONDecodeError:
raw_items = re.split(r"[\s,;]+", raw)
else:
raw_items = [value]
out: list[int] = []
for item in raw_items:
try:
recipient_id = int(str(item).strip())
except Exception:
continue
if recipient_id not in out:
out.append(recipient_id)
return out
def day_bounds(report_date: date) -> tuple[datetime, datetime]:
start_local = datetime.combine(report_date, time.min, tzinfo=LOCAL_TZ)
end_local = start_local + timedelta(days=1)
return start_local.astimezone(timezone.utc), end_local.astimezone(timezone.utc)
def fmt_int(value: Any) -> str:
try:
return f"{int(value):,}".replace(",", " ")
except Exception:
return "0"
def split_chunks(text: str, limit: int = MESSAGE_LIMIT) -> list[str]:
text = str(text or "").strip()
if not text:
return []
chunks: list[str] = []
while len(text) > limit:
split_at = text.rfind("\n", 0, limit)
if split_at < limit // 2:
split_at = limit
chunks.append(text[:split_at].strip())
text = text[split_at:].strip()
if text:
chunks.append(text)
return chunks
class DailyReportWorker:
def __init__(self) -> None:
self.pool = None
self.bot: Bot | None = None
self.bot_token = ""
self.recipients: list[int] = []
self.report_time = "00:04"
self.interval_sec = 60
self.heartbeat = HeartbeatReporter(WORKER_DAILY_REPORT, 30)
async def init(self) -> None:
self.pool = await get_pool()
await self.reload_settings()
async def close(self) -> None:
if self.bot:
await self.bot.session.close()
async def reload_settings(self) -> None:
token = (
str(await fetch_setting("daily_report_bot_token", "") or "").strip()
or str(await fetch_setting("tg_poster_bot_token", "") or "").strip()
or settings.tg_bot_token
)
if not token:
raise RuntimeError("daily_report_bot_token, tg_poster_bot_token and TG_BOT_TOKEN are empty")
if token != self.bot_token:
if self.bot:
await self.bot.session.close()
self.bot = Bot(token=token)
self.bot_token = token
self.recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142]))
if not self.recipients:
raise RuntimeError("daily_report_recipient_ids is empty")
report_time = str(await fetch_setting("daily_report_time", "00:04") or "00:04").strip()
self.report_time = report_time if valid_time(report_time) else "00:04"
self.interval_sec = max(10, await fetch_int_setting("daily_report_interval_sec", 60))
logger.info("Daily report config: time={} recipients={}", self.report_time, self.recipients)
async def save_last_sent_date(self, report_date: date) -> None:
await self.pool.execute(
"""
UPDATE app_settings
SET value_json=$2::jsonb,
updated_at=NOW()
WHERE key=$1
""",
"daily_report_last_sent_date",
json.dumps(report_date.isoformat()),
)
async def countval(self, sql: str, *args: Any) -> int:
return int(await self.pool.fetchval(sql, *args) or 0)
async def collect(self, report_date: date) -> dict[str, Any]:
start_utc, end_utc = day_bounds(report_date)
now_utc = datetime.now(timezone.utc)
async with self.pool.acquire() as conn:
raw_collected = await conn.fetchval(
"SELECT COUNT(*) FROM raw_posts WHERE created_at >= $1 AND created_at < $2",
start_utc,
end_utc,
)
storage_ready = await conn.fetchval(
"SELECT COUNT(*) FROM raw_posts WHERE status='storage_ready' AND updated_at >= $1 AND updated_at < $2",
start_utc,
end_utc,
)
media_failed = await conn.fetchval(
"SELECT COUNT(*) FROM raw_post_media WHERE status IN ('failed','link_only') AND updated_at >= $1 AND updated_at < $2",
start_utc,
end_utc,
)
qualification_rows = await conn.fetch(
"""
SELECT COALESCE(qualification_status, 'pending') AS status, COUNT(*) AS count
FROM raw_posts
WHERE qualified_at >= $1 AND qualified_at < $2
GROUP BY COALESCE(qualification_status, 'pending')
""",
start_utc,
end_utc,
)
writer_ready = await conn.fetchval(
"SELECT COUNT(*) FROM raw_posts WHERE rewrite_status='ready' AND rewritten_at >= $1 AND rewritten_at < $2",
start_utc,
end_utc,
)
review_now = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts
WHERE rewrite_status='ready'
AND COALESCE(editorial_status, 'review')='review'
"""
)
accepted_today = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts
WHERE reviewed_at >= $1 AND reviewed_at < $2
AND editorial_status='accepted'
""",
start_utc,
end_utc,
)
rejected_today = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts
WHERE reviewed_at >= $1 AND reviewed_at < $2
AND editorial_status='rejected'
""",
start_utc,
end_utc,
)
publication_rows = await conn.fetch(
"""
SELECT platform, status, COUNT(*) AS count
FROM post_publications
WHERE updated_at >= $1 AND updated_at < $2
GROUP BY platform, status
ORDER BY platform, status
""",
start_utc,
end_utc,
)
tg_queue = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts rp
WHERE rp.status='storage_ready'
AND rp.rewrite_status='ready'
AND rp.editorial_status='accepted'
AND COALESCE(rp.publication_status, 'pending')='pending'
AND EXISTS (
SELECT 1 FROM raw_post_media m
WHERE m.raw_post_id=rp.id
AND COALESCE(m.editor_hidden,FALSE)=FALSE
AND m.media_type IN ('photo','video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1 FROM raw_post_media m
WHERE m.raw_post_id=rp.id
AND COALESCE(m.editor_hidden,FALSE)=FALSE
AND m.media_type IN ('photo','video')
AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
)
"""
)
vk_queue = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts rp
LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='vk'
WHERE rp.status='storage_ready'
AND rp.rewrite_status='ready'
AND COALESCE(rp.editorial_status, 'review') IN ('accepted', 'published', 'publish_failed')
AND COALESCE(pp.status, 'pending') IN ('pending', 'publish_failed')
AND EXISTS (
SELECT 1 FROM raw_post_media rpm
WHERE rpm.raw_post_id=rp.id
AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
AND COALESCE(rpm.editor_added, FALSE)=FALSE
AND (
(rpm.media_type='photo' AND rpm.original_attachment_id LIKE 'photo%')
OR (rpm.media_type='video' AND rpm.original_attachment_id LIKE 'video%')
)
)
"""
)
publish_failed_now = await conn.fetchval(
"""
SELECT COUNT(*)
FROM raw_posts
WHERE rewrite_status='ready'
AND COALESCE(editorial_status, '')='publish_failed'
"""
)
workers = await conn.fetch(
"""
SELECT wc.name, wc.enabled, wh.heartbeat_at, wh.status
FROM worker_controls wc
LEFT JOIN worker_heartbeats wh ON wh.name=wc.name
ORDER BY wc.name
"""
)
top_errors = await conn.fetch(
"""
SELECT platform, error, COUNT(*) AS count
FROM post_publications
WHERE status='publish_failed'
AND updated_at >= $1 AND updated_at < $2
AND COALESCE(error, '') <> ''
GROUP BY platform, error
ORDER BY COUNT(*) DESC
LIMIT 5
""",
start_utc,
end_utc,
)
return {
"date": report_date,
"raw_collected": raw_collected,
"storage_ready": storage_ready,
"media_failed": media_failed,
"qualification": {row["status"]: int(row["count"]) for row in qualification_rows},
"writer_ready": writer_ready,
"review_now": review_now,
"accepted_today": accepted_today,
"rejected_today": rejected_today,
"publications": {(row["platform"], row["status"]): int(row["count"]) for row in publication_rows},
"tg_queue": tg_queue,
"vk_queue": vk_queue,
"publish_failed_now": publish_failed_now,
"workers": [dict(row) for row in workers],
"top_errors": [dict(row) for row in top_errors],
"now_utc": now_utc,
}
def render(self, data: dict[str, Any]) -> str:
q = data["qualification"]
pubs = data["publications"]
workers_lines: list[str] = []
for row in data["workers"]:
heartbeat_at = row.get("heartbeat_at")
is_stale = True
if heartbeat_at:
is_stale = (data["now_utc"] - heartbeat_at).total_seconds() > 600
marker = "OK" if row.get("enabled") and heartbeat_at and not is_stale else ("OFF" if not row.get("enabled") else "STALE")
workers_lines.append(f"{marker} {row['name']}: {row.get('status') or '-'}")
errors_lines = [
f"- {row['platform']}: {row['count']} x {str(row['error'])[:120]}"
for row in data["top_errors"]
]
if not errors_lines:
errors_lines = ["- нет"]
report_date = data["date"].strftime("%d.%m.%Y")
return "\n".join(
[
f"Ежедневный отчет N8 Parser за {report_date}",
"",
"Сбор:",
f"raw собрано: {fmt_int(data['raw_collected'])}",
f"storage ready за день: {fmt_int(data['storage_ready'])}",
f"media failed/link_only: {fmt_int(data['media_failed'])}",
"",
"AI:",
f"принято категоризатором: {fmt_int(q.get('accepted', 0))}",
f"отклонено категоризатором: {fmt_int(q.get('rejected', 0))}",
f"ошибки категоризатора: {fmt_int(q.get('failed', 0))}",
f"рерайтов готово: {fmt_int(data['writer_ready'])}",
"",
"Редакторская:",
f"принято редактором за день: {fmt_int(data['accepted_today'])}",
f"отклонено редактором за день: {fmt_int(data['rejected_today'])}",
f"на проверке сейчас: {fmt_int(data['review_now'])}",
f"ошибка публикации сейчас: {fmt_int(data['publish_failed_now'])}",
"",
"Публикации за день:",
f"TG опубликовано: {fmt_int(pubs.get(('tg', 'published'), 0))}",
f"VK опубликовано: {fmt_int(pubs.get(('vk', 'published'), 0))}",
f"TG ошибки: {fmt_int(pubs.get(('tg', 'publish_failed'), 0))}",
f"VK ошибки: {fmt_int(pubs.get(('vk', 'publish_failed'), 0))}",
"",
"Очередь сейчас:",
f"TG кандидатов: {fmt_int(data['tg_queue'])}",
f"VK кандидатов: {fmt_int(data['vk_queue'])}",
"",
"Ошибки дня:",
*errors_lines,
"",
"Воркеры:",
*workers_lines,
]
)
async def send_text(self, text: str) -> None:
if not self.bot:
raise RuntimeError("bot is not initialized")
for recipient_id in self.recipients:
for chunk in split_chunks(text):
while True:
try:
await self.bot.send_message(recipient_id, chunk, disable_web_page_preview=True)
break
except TelegramRetryAfter as exc:
await asyncio.sleep(float(exc.retry_after) + 1)
async def send_report(self, report_date: date, save_last_sent: bool = False) -> str:
await self.reload_settings()
data = await self.collect(report_date)
text = self.render(data)
await self.send_text(text)
if save_last_sent:
await self.save_last_sent_date(report_date)
await self.heartbeat.beat(
self.pool,
status="sent",
meta={"date": report_date.isoformat(), "recipients": self.recipients},
force=True,
)
return text
async def due_report_date(self) -> date | None:
now = datetime.now(LOCAL_TZ)
hour, minute = [int(part) for part in self.report_time.split(":", 1)]
slot_start = datetime.combine(now.date(), time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
slot_end = slot_start + timedelta(minutes=10)
if not (slot_start <= now < slot_end):
return None
report_date = now.date() - timedelta(days=1)
last_sent = str(await fetch_setting("daily_report_last_sent_date", "") or "").strip()
if last_sent == report_date.isoformat():
return None
return report_date
async def run_once(self) -> bool:
await self.reload_settings()
enabled = await is_worker_enabled(self.pool, WORKER_DAILY_REPORT)
app_enabled = await fetch_bool_setting("daily_report_enabled", True)
if not enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
report_date = await self.due_report_date()
if not report_date:
await self.heartbeat.beat(self.pool, status="idle")
return False
await self.send_report(report_date, save_last_sent=True)
return True
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_DAILY_REPORT, os.getpid())
while True:
try:
await self.run_once()
except Exception as exc:
logger.exception("Daily report loop error: {}", exc)
if self.pool:
await self.heartbeat.beat(self.pool, status="error", meta={"error": str(exc)}, force=True)
await asyncio.sleep(self.interval_sec)
async def run_send_now(report_date: date | None = None) -> str:
worker = DailyReportWorker()
await worker.init()
try:
return await worker.send_report(report_date or datetime.now(LOCAL_TZ).date(), save_last_sent=False)
finally:
await worker.close()
async def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--send-now", action="store_true")
parser.add_argument("--date", help="Report date YYYY-MM-DD, default today in EKB for --send-now")
args = parser.parse_args()
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
if args.send_now:
report_date = date.fromisoformat(args.date) if args.date else None
text = await run_send_now(report_date)
print(text)
return
worker = DailyReportWorker()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
+713
View File
@@ -0,0 +1,713 @@
from __future__ import annotations
import asyncio
import json
import mimetypes
import os
import re
import shutil
import sys
import tempfile
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import quote
from zoneinfo import ZoneInfo
import aiohttp
from loguru import logger
from ..config import settings
from ..constants import (
PUBLICATION_STATUS_FAILED,
PUBLICATION_STATUS_PENDING,
PUBLICATION_STATUS_PUBLISHED,
WORKER_MAX_POSTER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import build_publication_text
LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MAX_MESSAGE_LIMIT = 4000
MAX_MEDIA_ITEMS = 10
MAX_VIDEO_BYTES = 250 * 1024 * 1024
PROJECT_ROOT = Path(__file__).resolve().parents[3]
UPLOAD_ROOT = PROJECT_ROOT / "uploads"
class MAXAPIError(RuntimeError):
def __init__(self, message: str, status: int | None = None, code: str = "") -> None:
super().__init__(message)
self.status = status
self.code = code
def valid_time(value: str) -> bool:
return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))
def normalize_schedule(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list):
value = []
rows: list[dict[str, Any]] = []
for idx, item in enumerate(value):
if not isinstance(item, dict):
continue
time_value = str(item.get("time") or "").strip()
if not valid_time(time_value):
continue
count = max(0, min(20, int(item.get("count") or 0)))
if count < 1:
continue
rows.append(
{
"id": str(item.get("id") or f"slot-{idx + 1}").strip()[:80],
"time": time_value,
"count": count,
"enabled": bool(item.get("enabled", True)),
}
)
return sorted(rows, key=lambda row: row["time"])
def split_message_chunks(text: str, limit: int) -> list[str]:
text = str(text or "").strip()
if not text:
return []
chunks: list[str] = []
while len(text) > limit:
split_at = text.rfind("\n", 0, limit)
if split_at < limit // 2:
split_at = text.rfind(" ", 0, limit)
if split_at < limit // 2:
split_at = limit
chunks.append(text[:split_at].strip())
text = text[split_at:].strip()
if text:
chunks.append(text)
return chunks
def local_upload_path(url: str) -> Path | None:
raw = str(url or "").strip()
prefix = "/uploads/"
if not raw.startswith(prefix):
return None
path = (UPLOAD_ROOT / raw.removeprefix(prefix)).resolve()
try:
path.relative_to(UPLOAD_ROOT.resolve())
except ValueError:
return None
return path if path.is_file() else None
class MAXAPIClient:
def __init__(self, token: str, base_url: str, timeout_sec: int, max_attempts: int, retry_backoff_max_sec: int) -> None:
self.token = token
self.base_url = base_url.rstrip("/")
self.timeout_sec = timeout_sec
self.max_attempts = max_attempts
self.retry_backoff_max_sec = retry_backoff_max_sec
self.session: aiohttp.ClientSession | None = None
async def __aenter__(self) -> "MAXAPIClient":
timeout = aiohttp.ClientTimeout(total=self.timeout_sec)
self.session = aiohttp.ClientSession(timeout=timeout, headers={"Authorization": self.token})
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
if self.session:
await self.session.close()
async def request(self, method: str, path: str, **kwargs) -> dict[str, Any]:
if not self.session:
raise RuntimeError("MAX session is not initialized")
url = f"{self.base_url}{path}"
last_error = ""
for attempt in range(1, self.max_attempts + 1):
try:
async with self.session.request(method, url, **kwargs) as resp:
text = await resp.text()
try:
data = json.loads(text) if text else {}
except json.JSONDecodeError:
data = {"raw": text}
if 200 <= resp.status < 300:
return data
retry_after = resp.headers.get("Retry-After")
message = data.get("message") if isinstance(data, dict) else text
code = data.get("code") if isinstance(data, dict) else ""
last_error = f"MAX API {resp.status} {code}: {message or text}"
if resp.status == 429 and retry_after:
await asyncio.sleep(float(retry_after) + 0.5)
continue
if code == "attachment.not.ready" and attempt < self.max_attempts:
await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec))
continue
if resp.status < 500:
raise MAXAPIError(last_error, resp.status, str(code or ""))
except Exception as exc:
last_error = str(exc)
if attempt >= self.max_attempts:
raise
await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec))
raise RuntimeError(last_error or "MAX API request failed")
async def send_message(self, chat_id: int, text: str, attachments: list[dict[str, Any]] | None = None) -> dict[str, Any]:
payload: dict[str, Any] = {"text": text, "notify": True}
if attachments:
payload["attachments"] = attachments
return await self.request("POST", f"/messages?chat_id={chat_id}", json=payload)
async def get_video_info(self, token: str) -> dict[str, Any]:
return await self.request("GET", f"/videos/{quote(token, safe='')}")
async def get_upload_url(self, media_type: str) -> dict[str, Any]:
upload_type = "video" if media_type == "video" else "image"
return await self.request("POST", f"/uploads?type={upload_type}")
async def upload_file(self, upload_url: str, path: Path) -> dict[str, Any]:
if not self.session:
raise RuntimeError("MAX session is not initialized")
form = aiohttp.FormData()
content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
with path.open("rb") as fh:
form.add_field("data", fh, filename=path.name, content_type=content_type)
async with self.session.post(upload_url, data=form) as resp:
text = await resp.text()
try:
data = json.loads(text) if text else {}
except json.JSONDecodeError:
data = {"raw": text}
if resp.status < 200 or resp.status >= 300:
raise RuntimeError(f"MAX upload {resp.status}: {data}")
return data
async def react(self, path_template: str, message_id: str, reaction: str) -> None:
path = path_template.format(message_id=message_id)
payload = {"reaction": reaction}
await self.request("POST", path, json=payload)
class MAXPoster:
def __init__(self) -> None:
self.pool = None
self.heartbeat = HeartbeatReporter(WORKER_MAX_POSTER, 30)
self.token = ""
self.chat_id = 0
self.api_base_url = "https://platform-api2.max.ru"
self.schedule: list[dict[str, Any]] = []
self.interval_sec = 60
self.message_limit = MAX_MESSAGE_LIMIT
self.media_max_items = MAX_MEDIA_ITEMS
self.media_process_delay_sec = 3.0
self.video_ready_attempts = 5
self.video_ready_delay_sec = 10.0
self.max_attempts = 3
self.retry_backoff_max_sec = 30
self.send_delay_sec = 1.0
self.recent_window = 20
self.category_repeat_penalty = 3.0
self.source_repeat_penalty = 4.0
self.auto_reaction_enabled = True
self.auto_reaction = "👍"
self.reaction_path_template = "/messages/{message_id}/reactions"
self.dry_run = False
async def init(self) -> None:
self.pool = await get_pool()
async def reload_settings(self) -> None:
self.token = str(await fetch_setting("max_poster_bot_token", "") or "").strip()
self.chat_id = int(str(await fetch_setting("max_poster_chat_id", "0") or "0").strip())
self.api_base_url = str(await fetch_setting("max_poster_api_base_url", self.api_base_url) or self.api_base_url).strip()
self.schedule = normalize_schedule(await fetch_setting("max_poster_schedule_json", []))
self.interval_sec = max(10, await fetch_int_setting("max_poster_interval_sec", 60))
self.message_limit = max(512, min(MAX_MESSAGE_LIMIT, await fetch_int_setting("max_poster_message_limit", MAX_MESSAGE_LIMIT)))
self.media_max_items = max(0, min(MAX_MEDIA_ITEMS, await fetch_int_setting("max_poster_media_max_items", MAX_MEDIA_ITEMS)))
self.media_process_delay_sec = max(0.0, await fetch_float_setting("max_poster_media_process_delay_sec", 3.0))
self.video_ready_attempts = max(1, await fetch_int_setting("max_poster_video_ready_attempts", 5))
self.video_ready_delay_sec = max(1.0, await fetch_float_setting("max_poster_video_ready_delay_sec", 10.0))
self.max_attempts = max(1, await fetch_int_setting("max_poster_max_attempts", 3))
self.retry_backoff_max_sec = max(1, await fetch_int_setting("max_poster_retry_backoff_max_sec", 30))
self.send_delay_sec = max(0.0, await fetch_float_setting("max_poster_send_delay_sec", 1.0))
self.recent_window = max(1, await fetch_int_setting("max_poster_recent_window", 20))
self.category_repeat_penalty = max(0.0, await fetch_float_setting("max_poster_category_repeat_penalty", 3.0))
self.source_repeat_penalty = max(0.0, await fetch_float_setting("max_poster_source_repeat_penalty", 4.0))
self.auto_reaction_enabled = await fetch_bool_setting("max_poster_auto_reaction_enabled", True)
self.auto_reaction = str(await fetch_setting("max_poster_auto_reaction", "👍") or "👍").strip() or "👍"
self.reaction_path_template = str(await fetch_setting("max_poster_reaction_path_template", self.reaction_path_template) or "").strip()
self.dry_run = await fetch_bool_setting("max_poster_dry_run", False)
if not self.token:
raise RuntimeError("max_poster_bot_token is empty")
if not self.chat_id:
raise RuntimeError("max_poster_chat_id is empty")
logger.info("MAX poster config: chat={} schedule={} dry_run={}", self.chat_id, self.schedule, self.dry_run)
async def due_slots(self) -> list[dict[str, Any]]:
now = datetime.now(LOCAL_TZ)
today = now.date()
due: list[dict[str, Any]] = []
for row in self.schedule:
if not row.get("enabled"):
continue
hour, minute = [int(part) for part in str(row["time"]).split(":", 1)]
slot_start = datetime.combine(today, time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
slot_end = slot_start + timedelta(minutes=10)
if not (slot_start <= now < slot_end):
continue
inserted = await self.pool.fetchrow(
"""
INSERT INTO publication_runs(poster, schedule_id, scheduled_for, scheduled_time, planned_count, status)
VALUES($1, $2, $3, $4, $5, 'started')
ON CONFLICT (poster, schedule_id, scheduled_for) DO NOTHING
RETURNING id
""",
"max",
row["id"],
today,
row["time"],
int(row["count"]),
)
if inserted:
due.append({**row, "run_id": int(inserted["id"]), "date": today})
return due
async def load_recent(self) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT rp.final_category_tag, rp.rewrite_category_tag, rp.final_source_tag, rp.rewrite_source_tag
FROM post_publications pp
JOIN raw_posts rp ON rp.id=pp.raw_post_id
WHERE pp.platform='max'
AND pp.status=$1
AND pp.target_id=$2
ORDER BY pp.published_at DESC NULLS LAST, pp.id DESC
LIMIT $3::int
""",
PUBLICATION_STATUS_PUBLISHED,
str(self.chat_id),
self.recent_window,
)
return [dict(row) for row in rows]
async def load_candidates(self, limit: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT rp.*, s.name AS source_name, s.tag AS source_tag
FROM raw_posts rp
JOIN sources s ON s.id=rp.source_id
LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='max'
WHERE rp.status='storage_ready'
AND rp.rewrite_status='ready'
AND COALESCE(rp.editorial_status, 'review') IN ('accepted', 'published', 'publish_failed')
AND COALESCE(pp.status, $1) = $1
AND EXISTS (
SELECT 1
FROM raw_post_media rpm
WHERE rpm.raw_post_id=rp.id
AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
AND rpm.media_type IN ('photo', 'video')
AND COALESCE(rpm.original_url, '') <> ''
)
ORDER BY COALESCE(rp.reviewed_at, rp.edited_at, rp.rewritten_at, rp.created_at) ASC,
rp.id ASC
LIMIT $2::int
""",
PUBLICATION_STATUS_PENDING,
max(limit, 50),
)
return [dict(row) for row in rows]
def choose_posts(self, candidates: list[dict[str, Any]], recent: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
selected: list[dict[str, Any]] = []
recent_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in recent]
recent_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or "") for row in recent]
pool = list(candidates)
while pool and len(selected) < count:
best_idx = 0
selected_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in selected]
selected_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or row.get("source_tag") or "") for row in selected]
previous_category = selected_categories[-1] if selected_categories else (recent_categories[0] if recent_categories else "")
previous_source = selected_sources[-1] if selected_sources else (recent_sources[0] if recent_sources else "")
best_rank: tuple[int, float, int] | None = None
for idx, post in enumerate(pool):
category = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or "")
source = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
immediate_repeats = int(bool(category) and category == previous_category) + int(bool(source) and source == previous_source)
repeat_penalty = (
(recent_categories.count(category) + selected_categories.count(category)) * self.category_repeat_penalty
+ (recent_sources.count(source) + selected_sources.count(source)) * self.source_repeat_penalty
)
rank = (immediate_repeats, repeat_penalty, idx)
if best_rank is None or rank < best_rank:
best_idx = idx
best_rank = rank
selected.append(pool.pop(best_idx))
return selected
async def load_media(self, raw_post_id: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT *
FROM raw_post_media
WHERE raw_post_id=$1
AND COALESCE(editor_hidden, FALSE)=FALSE
AND media_type IN ('photo', 'video')
AND COALESCE(original_url, '') <> ''
ORDER BY sort_order ASC, id ASC
LIMIT $2::int
""",
raw_post_id,
self.media_max_items,
)
return [dict(row) for row in rows]
def build_text(self, post: dict[str, Any]) -> str:
category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag)
async def download_media_to_temp(self, session: aiohttp.ClientSession, item: dict[str, Any]) -> Path | None:
media_type = str(item.get("media_type") or "")
file_id = str(item.get("tg_file_id") or item.get("storage_attachment_id") or "").strip()
if media_type == "video" and file_id:
return await self.download_telegram_file_to_temp(session, file_id, ".mp4")
url = str(item.get("original_url") or "").strip()
local_path = local_upload_path(url)
if local_path:
return local_path
if not url.startswith(("http://", "https://")):
return None
suffix = Path(url.split("?", 1)[0]).suffix[:12] or (".mp4" if item.get("media_type") == "video" else ".jpg")
tmp = tempfile.NamedTemporaryFile(prefix="max-poster-", suffix=suffix, delete=False)
path = Path(tmp.name)
tmp.close()
total = 0
try:
async with session.get(url) as resp:
resp.raise_for_status()
with path.open("wb") as fh:
async for chunk in resp.content.iter_chunked(1024 * 512):
total += len(chunk)
if total > MAX_VIDEO_BYTES:
raise RuntimeError("media file is larger than MAX video limit")
fh.write(chunk)
return path
except Exception:
path.unlink(missing_ok=True)
raise
async def download_telegram_file_to_temp(self, session: aiohttp.ClientSession, file_id: str, suffix: str) -> Path:
token = settings.tg_bot_token
if not token:
raise RuntimeError("TG_BOT_TOKEN is empty; cannot download stored video")
api_base = str(await fetch_setting("local_bot_api_url", settings.local_bot_api_url) or "").strip().rstrip("/")
if not api_base:
api_base = "https://api.telegram.org"
get_file_url = f"{api_base}/bot{token}/getFile"
async with session.post(get_file_url, json={"file_id": file_id}) as resp:
data = await resp.json(content_type=None)
if not data.get("ok") or not isinstance(data.get("result"), dict):
raise RuntimeError(f"Telegram getFile failed: {data}")
file_path = str(data["result"].get("file_path") or "").strip()
if not file_path:
raise RuntimeError(f"Telegram getFile did not return file_path: {data}")
local_file = Path(file_path)
if local_file.is_absolute() and local_file.is_file():
if local_file.stat().st_size > MAX_VIDEO_BYTES:
raise RuntimeError("telegram local video file is larger than MAX video limit")
tmp = tempfile.NamedTemporaryFile(prefix="max-poster-tg-", suffix=suffix, delete=False)
path = Path(tmp.name)
tmp.close()
try:
shutil.copyfile(local_file, path)
return path
except Exception:
path.unlink(missing_ok=True)
raise
download_url = f"{api_base}/file/bot{token}/{file_path}"
tmp = tempfile.NamedTemporaryFile(prefix="max-poster-tg-", suffix=suffix, delete=False)
path = Path(tmp.name)
tmp.close()
total = 0
try:
async with session.get(download_url) as resp:
resp.raise_for_status()
with path.open("wb") as fh:
async for chunk in resp.content.iter_chunked(1024 * 1024):
total += len(chunk)
if total > MAX_VIDEO_BYTES:
raise RuntimeError("telegram video file is larger than MAX video limit")
fh.write(chunk)
return path
except Exception:
path.unlink(missing_ok=True)
raise
async def upload_media(self, client: MAXAPIClient, item: dict[str, Any]) -> dict[str, Any] | None:
media_type = str(item.get("media_type") or "")
if media_type not in {"photo", "video"}:
return None
temp_path: Path | None = None
downloaded = False
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=180)) as session:
temp_path = await self.download_media_to_temp(session, item)
downloaded = bool(temp_path and not local_upload_path(str(item.get("original_url") or "")))
if not temp_path:
return None
try:
upload_info = await client.get_upload_url("video" if media_type == "video" else "image")
upload_url = str(upload_info.get("url") or "").strip()
if not upload_url:
raise RuntimeError(f"MAX upload URL is empty: {upload_info}")
payload = await client.upload_file(upload_url, temp_path)
if media_type == "video" and upload_info.get("token") and not payload.get("token"):
payload["token"] = upload_info["token"]
has_image_token = bool(payload.get("photos")) if isinstance(payload.get("photos"), dict) else False
if not payload.get("token") and not has_image_token:
raise RuntimeError(f"MAX upload did not return token: {payload}")
return {"type": "video" if media_type == "video" else "image", "payload": payload}
finally:
if downloaded and temp_path:
temp_path.unlink(missing_ok=True)
def message_id_from_response(self, data: dict[str, Any]) -> str:
candidates = [
data.get("id"),
data.get("mid"),
(data.get("message") or {}).get("id") if isinstance(data.get("message"), dict) else None,
(data.get("message") or {}).get("mid") if isinstance(data.get("message"), dict) else None,
]
message = data.get("message") if isinstance(data.get("message"), dict) else {}
body = message.get("body") if isinstance(message.get("body"), dict) else {}
candidates.extend([body.get("id"), body.get("mid"), body.get("message_id")])
return next((str(value) for value in candidates if value), "")
async def send_message_waiting_for_media(
self,
client: MAXAPIClient,
text: str,
attachments: list[dict[str, Any]],
) -> dict[str, Any]:
has_video = any(item.get("type") == "video" for item in attachments)
attempts = self.video_ready_attempts if has_video else 1
delay = self.video_ready_delay_sec
for attempt in range(1, attempts + 1):
try:
return await client.send_message(self.chat_id, text, attachments=attachments or None)
except MAXAPIError as exc:
if exc.code != "attachment.not.ready" or attempt >= attempts:
raise
logger.warning(
"MAX attachment is not ready, retry send in {}s attempt={}/{}",
delay,
attempt + 1,
attempts,
)
await asyncio.sleep(delay)
delay = min(delay * 2, self.retry_backoff_max_sec * 4)
raise RuntimeError("MAX attachment readiness retry exhausted")
def video_tokens(self, attachments: list[dict[str, Any]]) -> list[str]:
tokens: list[str] = []
for item in attachments:
if item.get("type") != "video":
continue
payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
token = str(payload.get("token") or "").strip()
if token:
tokens.append(token)
return tokens
async def wait_for_videos(self, client: MAXAPIClient, attachments: list[dict[str, Any]]) -> None:
tokens = self.video_tokens(attachments)
if not tokens:
return
delay = self.video_ready_delay_sec
for attempt in range(1, self.video_ready_attempts + 1):
pending: list[str] = []
for token in tokens:
info = await client.get_video_info(token)
if not info.get("urls"):
pending.append(token)
if not pending:
return
if attempt >= self.video_ready_attempts:
raise RuntimeError(f"MAX video is not ready after {attempt} attempts")
logger.warning(
"MAX video is not ready, retry info in {}s attempt={}/{} pending={}",
delay,
attempt + 1,
self.video_ready_attempts,
len(pending),
)
await asyncio.sleep(delay)
delay = min(delay * 2, self.retry_backoff_max_sec * 4)
async def send_post(self, post: dict[str, Any]) -> tuple[list[str], str | None, list[dict[str, Any]], str | None]:
text = self.build_text(post)
chunks = split_message_chunks(text, self.message_limit)
media_rows = await self.load_media(int(post["id"])) if self.media_max_items else []
message_ids: list[str] = []
attachments: list[dict[str, Any]] = []
first_url: str | None = None
reaction_error: str | None = None
async with MAXAPIClient(
self.token,
self.api_base_url,
timeout_sec=max(30, await fetch_int_setting("max_poster_timeout_sec", 120)),
max_attempts=self.max_attempts,
retry_backoff_max_sec=self.retry_backoff_max_sec,
) as client:
for item in media_rows:
try:
attachment = await self.upload_media(client, item)
if attachment:
attachments.append(attachment)
except Exception as exc:
logger.warning("MAX media skipped raw_post={} media_id={}: {}", post["id"], item.get("id"), exc)
if len(attachments) >= self.media_max_items:
break
if self.media_process_delay_sec and attachments:
await asyncio.sleep(self.media_process_delay_sec)
if self.dry_run:
return ["dry-run"], None, attachments, None
await self.wait_for_videos(client, attachments)
first_text = chunks[0] if chunks else ""
first = await self.send_message_waiting_for_media(client, first_text, attachments)
message = first.get("message") if isinstance(first.get("message"), dict) else first
first_url = message.get("url") if isinstance(message, dict) else None
first_message_id = self.message_id_from_response(first)
if first_message_id:
message_ids.append(first_message_id)
if self.auto_reaction_enabled and self.reaction_path_template:
try:
await client.react(self.reaction_path_template, first_message_id, self.auto_reaction)
except Exception as exc:
reaction_error = str(exc)
logger.warning("MAX auto reaction failed message={}: {}", first_message_id, exc)
for chunk in chunks[1:]:
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
result = await client.send_message(self.chat_id, chunk)
message_id = self.message_id_from_response(result)
if message_id:
message_ids.append(message_id)
return message_ids, first_url, attachments, reaction_error
async def upsert_publication(
self,
post_id: int,
status: str,
external_id: str | None = None,
url: str | None = None,
error: str | None = None,
attachments: list[dict[str, Any]] | None = None,
) -> None:
await self.pool.execute(
"""
INSERT INTO post_publications(
raw_post_id, platform, status, target_id, external_id, url, error, attachments_json, published_at, updated_at
)
VALUES($1, 'max', $2, $3, $4, $5, $6, $7::jsonb, CASE WHEN $2=$8 THEN NOW() ELSE NULL END, NOW())
ON CONFLICT (raw_post_id, platform) DO UPDATE
SET status=EXCLUDED.status,
target_id=EXCLUDED.target_id,
external_id=EXCLUDED.external_id,
url=EXCLUDED.url,
error=EXCLUDED.error,
attachments_json=EXCLUDED.attachments_json,
published_at=CASE WHEN EXCLUDED.status=$8 THEN COALESCE(EXCLUDED.published_at, NOW()) ELSE post_publications.published_at END,
updated_at=NOW()
""",
post_id,
status,
str(self.chat_id),
external_id,
url,
error[:1000] if error else None,
json.dumps(attachments or [], ensure_ascii=False),
PUBLICATION_STATUS_PUBLISHED,
)
async def process_slot(self, slot: dict[str, Any]) -> int:
count = int(slot["count"])
recent = await self.load_recent()
candidates = await self.load_candidates(max(50, count * 10))
posts = self.choose_posts(candidates, recent, count)
published = 0
for post in posts:
await self.heartbeat.beat(self.pool, status="publishing", meta={"post_id": int(post["id"]), "slot": slot["id"]}, force=True)
try:
message_ids, url, attachments, reaction_error = await self.send_post(post)
await self.upsert_publication(
int(post["id"]),
PUBLICATION_STATUS_PUBLISHED,
",".join(message_ids),
url,
error=None,
attachments=attachments,
)
published += 1
logger.info("MAX published raw_post={} messages={} attachments={}", post["id"], message_ids, len(attachments))
except Exception as exc:
await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_FAILED, error=str(exc))
logger.exception("MAX publish failed raw_post={}: {}", post["id"], exc)
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
await self.pool.execute(
"""
UPDATE publication_runs
SET published_count=$2,
status=$3,
error=$4,
updated_at=NOW()
WHERE id=$1
""",
int(slot["run_id"]),
published,
"done" if published == count else "partial",
None if published == count else f"published {published} of {count}",
)
return published
async def run_once(self) -> bool:
await self.reload_settings()
enabled = await is_worker_enabled(self.pool, WORKER_MAX_POSTER)
app_enabled = await fetch_bool_setting("max_poster_enabled", False)
if not enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
slots = await self.due_slots()
if not slots:
await self.heartbeat.beat(self.pool, status="idle")
return False
total = 0
for slot in slots:
total += await self.process_slot(slot)
await self.heartbeat.beat(self.pool, status="idle", meta={"published": total}, force=True)
return bool(total)
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_MAX_POSTER, os.getpid())
while True:
try:
await self.run_once()
except Exception as exc:
logger.exception("MAX poster loop error: {}", exc)
await asyncio.sleep(self.interval_sec)
async def main() -> None:
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
worker = MAXPoster()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
+141
View File
@@ -0,0 +1,141 @@
from __future__ import annotations
import asyncio
import os
import sys
from pathlib import Path
from loguru import logger
from ..config import settings
from ..constants import WORKER_SITE_POSTER
from ..db import fetch_bool_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
class SitePoster:
def __init__(self) -> None:
self.pool = None
self.heartbeat = HeartbeatReporter(WORKER_SITE_POSTER, 30)
self.project_root = Path(__file__).resolve().parents[3]
self.interval_sec = 300
self.photo_batch_size = 10
self.video_batch_size = 1
self.ghost_url = "https://f-n8.ru"
self.key_file = "/opt/vk-parser/.site-poster-key"
async def init(self) -> None:
self.pool = await get_pool()
async def reload_settings(self) -> None:
self.interval_sec = max(30, await fetch_int_setting("site_poster_interval_sec", 300))
self.photo_batch_size = max(0, await fetch_int_setting("site_poster_photo_batch_size", 10))
self.video_batch_size = max(0, await fetch_int_setting("site_poster_video_batch_size", 1))
self.ghost_url = str(await fetch_setting("site_poster_ghost_url", "https://f-n8.ru") or "https://f-n8.ru").rstrip("/")
self.key_file = str(await fetch_setting("site_poster_key_file", "/opt/vk-parser/.site-poster-key") or "").strip()
if not self.key_file:
raise RuntimeError("site_poster_key_file is empty")
async def publish_mode(self, flag: str, limit: int) -> tuple[int, int]:
if limit < 1:
return 0, 0
command = [
sys.executable,
str(self.project_root / "scripts" / "publish_site_batch.py"),
flag,
"--limit",
str(limit),
"--ghost-url",
self.ghost_url,
"--key-file",
self.key_file,
]
env = os.environ.copy()
env["PYTHONPATH"] = str(self.project_root / "src")
env["PYTHONUNBUFFERED"] = "1"
process = await asyncio.create_subprocess_exec(
*command,
cwd=str(self.project_root),
env=env,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
published = 0
failed = 0
assert process.stdout is not None
while True:
try:
raw_line = await asyncio.wait_for(process.stdout.readline(), timeout=20)
except TimeoutError:
await self.heartbeat.beat(
self.pool,
status="publishing",
meta={"mode": flag, "published": published, "failed": failed},
force=True,
)
continue
if not raw_line:
break
line = raw_line.decode("utf-8", errors="replace").strip()
if not line:
continue
if line.startswith("PUBLISHED "):
published += 1
logger.info("Site poster: {}", line)
elif line.startswith("FAILED "):
failed += 1
logger.error("Site poster: {}", line)
elif line not in {"NO_READY_POSTS", "ALREADY_RUNNING"}:
logger.debug("Site poster child: {}", line)
await self.heartbeat.beat(
self.pool,
status="publishing",
meta={"mode": flag, "published": published, "failed": failed},
)
return_code = await process.wait()
if return_code != 0:
raise RuntimeError(f"site publisher exited with code {return_code} for {flag}")
return published, failed
async def run_once(self) -> bool:
await self.reload_settings()
worker_enabled = await is_worker_enabled(self.pool, WORKER_SITE_POSTER)
app_enabled = await fetch_bool_setting("site_poster_enabled", False)
if not worker_enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
await self.heartbeat.beat(self.pool, status="checking", force=True)
photo_published, photo_failed = await self.publish_mode("--all-ready", self.photo_batch_size)
video_published, video_failed = await self.publish_mode("--all-ready-videos", self.video_batch_size)
published = photo_published + video_published
failed = photo_failed + video_failed
await self.heartbeat.beat(
self.pool,
status="idle" if not failed else "error",
meta={"published": published, "failed": failed},
force=True,
)
return bool(published)
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_SITE_POSTER, os.getpid())
while True:
try:
await self.run_once()
except Exception as exc:
logger.exception("Site poster loop error: {}", exc)
await self.heartbeat.beat(self.pool, status="error", meta={"error": str(exc)[:500]}, force=True)
await asyncio.sleep(self.interval_sec)
async def main() -> None:
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
worker = SitePoster()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
+535
View File
@@ -0,0 +1,535 @@
from __future__ import annotations
import asyncio
import json
import os
import re
import sys
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
from aiogram import Bot
from aiogram.exceptions import TelegramRetryAfter
from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo
from loguru import logger
from ..config import settings
from ..constants import (
PUBLICATION_STATUS_FAILED,
PUBLICATION_STATUS_PENDING,
PUBLICATION_STATUS_PUBLISHED,
WORKER_TG_POSTER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import build_publication_text
LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MAX_MEDIA_GROUP = 10
PROJECT_ROOT = Path(__file__).resolve().parents[3]
UPLOAD_ROOT = PROJECT_ROOT / "uploads"
def parse_topic(value: str) -> tuple[int, int | None]:
raw = str(value or "").strip()
if not raw:
raise RuntimeError("tg_poster_chat_id is empty")
if ":" in raw:
chat_id, thread_id = raw.split(":", 1)
return int(chat_id), int(thread_id)
return int(raw), None
def valid_time(value: str) -> bool:
return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))
def split_message_chunks(text: str, limit: int) -> list[str]:
text = str(text or "").strip()
if not text:
return []
chunks: list[str] = []
while len(text) > limit:
split_at = text.rfind("\n", 0, limit)
if split_at < limit // 2:
split_at = text.rfind(" ", 0, limit)
if split_at < limit // 2:
split_at = limit
chunks.append(text[:split_at].strip())
text = text[split_at:].strip()
if text:
chunks.append(text)
return chunks
def tg_message_url(chat_id: int, message_id: int) -> str:
chat = str(abs(int(chat_id)))
if chat.startswith("100"):
chat = chat[3:]
return f"https://t.me/c/{chat}/{message_id}"
def local_upload_path(url: str) -> Path | None:
raw = str(url or "").strip()
prefix = "/uploads/"
if not raw.startswith(prefix):
return None
path = (UPLOAD_ROOT / raw.removeprefix(prefix)).resolve()
try:
path.relative_to(UPLOAD_ROOT.resolve())
except ValueError:
return None
return path if path.is_file() else None
class TelegramPoster:
def __init__(self) -> None:
self.pool = None
self.bot: Bot | None = None
self.bot_token = ""
self.chat_id: int | None = None
self.thread_id: int | None = None
self.heartbeat = HeartbeatReporter(WORKER_TG_POSTER, 30)
self.schedule: list[dict[str, Any]] = []
self.media_group_max_items = MAX_MEDIA_GROUP
self.caption_limit = 1024
self.message_limit = 4096
self.max_attempts = 3
self.retry_backoff_max_sec = 30
self.send_delay_sec = 1.0
self.overflow_caption = "⬇️ Описание"
self.recent_window = 20
self.category_repeat_penalty = 3.0
self.source_repeat_penalty = 4.0
async def init(self) -> None:
self.pool = await get_pool()
await self.reload_settings()
async def reload_settings(self) -> None:
token = str(await fetch_setting("tg_poster_bot_token", "") or "").strip() or settings.tg_bot_token
if not token:
raise RuntimeError("tg_poster_bot_token and TG_BOT_TOKEN are empty")
if token != self.bot_token:
if self.bot:
await self.bot.session.close()
self.bot = Bot(token=token)
self.bot_token = token
self.chat_id, self.thread_id = parse_topic(str(await fetch_setting("tg_poster_chat_id", "-1003712724327") or ""))
self.schedule = self.normalize_schedule(await fetch_setting("tg_poster_schedule_json", []))
self.media_group_max_items = max(1, min(MAX_MEDIA_GROUP, await fetch_int_setting("tg_poster_media_group_max_items", 10)))
self.caption_limit = max(128, min(1024, await fetch_int_setting("tg_poster_caption_limit", 1024)))
self.message_limit = max(512, min(4096, await fetch_int_setting("tg_poster_message_limit", 4096)))
self.max_attempts = max(1, await fetch_int_setting("tg_poster_max_attempts", 3))
self.retry_backoff_max_sec = max(1, await fetch_int_setting("tg_poster_retry_backoff_max_sec", 30))
self.send_delay_sec = max(0.0, await fetch_float_setting("tg_poster_send_delay_sec", 1.0))
self.overflow_caption = str(await fetch_setting("tg_poster_text_overflow_caption", self.overflow_caption) or self.overflow_caption)
self.recent_window = max(1, await fetch_int_setting("tg_poster_recent_window", 20))
self.category_repeat_penalty = max(0.0, await fetch_float_setting("tg_poster_category_repeat_penalty", 3.0))
self.source_repeat_penalty = max(0.0, await fetch_float_setting("tg_poster_source_repeat_penalty", 4.0))
logger.info("TG poster config: chat={} schedule={} caption_limit={}", self.chat_id, self.schedule, self.caption_limit)
async def close(self) -> None:
if self.bot:
await self.bot.session.close()
def normalize_schedule(self, value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list):
value = []
rows: list[dict[str, Any]] = []
for idx, item in enumerate(value):
if not isinstance(item, dict):
continue
time_value = str(item.get("time") or "").strip()
if not valid_time(time_value):
continue
count = max(0, min(20, int(item.get("count") or 0)))
if count < 1:
continue
rows.append(
{
"id": str(item.get("id") or f"slot-{idx + 1}").strip()[:80],
"time": time_value,
"count": count,
"enabled": bool(item.get("enabled", True)),
}
)
return sorted(rows, key=lambda row: row["time"])
def chat_kwargs(self) -> dict[str, int]:
kwargs = {"chat_id": int(self.chat_id)}
if self.thread_id:
kwargs["message_thread_id"] = int(self.thread_id)
return kwargs
async def tg_retry(self, fn):
for attempt in range(1, self.max_attempts + 1):
try:
return await fn()
except TelegramRetryAfter as exc:
delay = float(exc.retry_after) + 0.5
logger.warning("Telegram flood control, sleep {}s", delay)
await asyncio.sleep(delay)
except Exception as exc:
logger.warning("Telegram post error without retry to avoid duplicate send: {}", exc)
raise
raise RuntimeError("telegram retry exhausted")
async def due_slots(self) -> list[dict[str, Any]]:
now = datetime.now(LOCAL_TZ)
today = now.date()
due: list[dict[str, Any]] = []
for row in self.schedule:
if not row.get("enabled"):
continue
hour, minute = [int(part) for part in str(row["time"]).split(":", 1)]
slot_start = datetime.combine(today, time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
slot_end = slot_start + timedelta(minutes=10)
if not (slot_start <= now < slot_end):
continue
inserted = await self.pool.fetchrow(
"""
INSERT INTO publication_runs(poster, schedule_id, scheduled_for, scheduled_time, planned_count, status)
VALUES($1, $2, $3, $4, $5, 'started')
ON CONFLICT (poster, schedule_id, scheduled_for) DO NOTHING
RETURNING id
""",
"tg",
row["id"],
today,
row["time"],
int(row["count"]),
)
if inserted:
due.append({**row, "run_id": int(inserted["id"]), "date": today})
return due
async def load_recent(self) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT final_category_tag, rewrite_category_tag, final_source_tag, rewrite_source_tag
FROM raw_posts
WHERE publication_status=$1
AND tg_publication_chat_id=$2
ORDER BY published_at DESC NULLS LAST, id DESC
LIMIT $3::int
""",
PUBLICATION_STATUS_PUBLISHED,
self.chat_id,
self.recent_window,
)
return [dict(row) for row in rows]
async def load_candidates(self, limit: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT rp.*, s.name AS source_name, s.tag AS source_tag
FROM raw_posts rp
JOIN sources s ON s.id=rp.source_id
WHERE rp.status='storage_ready'
AND rp.rewrite_status='ready'
AND rp.editorial_status='accepted'
AND COALESCE(rp.publication_status, 'pending') = $1
AND EXISTS (
SELECT 1
FROM raw_post_media rpm
WHERE rpm.raw_post_id=rp.id
AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
AND rpm.media_type IN ('photo', 'video')
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 COALESCE(rpm.editor_hidden, FALSE)=FALSE
AND rpm.media_type IN ('photo', 'video')
AND COALESCE(rpm.tg_file_id, rpm.storage_attachment_id, '') = ''
)
ORDER BY COALESCE(rp.reviewed_at, rp.edited_at, rp.rewritten_at, rp.created_at) ASC,
rp.id ASC
LIMIT $2::int
""",
PUBLICATION_STATUS_PENDING,
max(limit, 50),
)
return [dict(row) for row in rows]
def choose_posts(self, candidates: list[dict[str, Any]], recent: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
selected: list[dict[str, Any]] = []
recent_categories = [
str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "")
for row in recent
]
recent_sources = [
str(row.get("final_source_tag") or row.get("rewrite_source_tag") or "")
for row in recent
]
pool = list(candidates)
while pool and len(selected) < count:
best_idx = 0
selected_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in selected]
selected_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or row.get("source_tag") or "") for row in selected]
previous_category = selected_categories[-1] if selected_categories else (recent_categories[0] if recent_categories else "")
previous_source = selected_sources[-1] if selected_sources else (recent_sources[0] if recent_sources else "")
best_rank: tuple[int, float, int] | None = None
for idx, post in enumerate(pool):
category = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or "")
source = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
immediate_repeats = int(bool(category) and category == previous_category) + int(bool(source) and source == previous_source)
repeat_penalty = (
(recent_categories.count(category) + selected_categories.count(category)) * self.category_repeat_penalty
+ (recent_sources.count(source) + selected_sources.count(source)) * self.source_repeat_penalty
)
rank = (immediate_repeats, repeat_penalty, idx)
if best_rank is None or rank < best_rank:
best_idx = idx
best_rank = rank
selected.append(pool.pop(best_idx))
return selected
async def load_media(self, raw_post_id: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT *
FROM raw_post_media
WHERE raw_post_id=$1
AND COALESCE(editor_hidden, FALSE)=FALSE
AND media_type IN ('photo', 'video')
ORDER BY sort_order ASC, id ASC
""",
raw_post_id,
)
return [dict(row) for row in rows]
def media_value(self, item: dict[str, Any]) -> Any | None:
stored = item.get("tg_file_id") or item.get("storage_attachment_id")
if stored:
return str(stored)
url = str(item.get("original_url") or "").strip()
path = local_upload_path(url)
if path:
return FSInputFile(path)
return None
def build_text(self, post: dict[str, Any]) -> str:
category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag)
async def send_text(self, text: str) -> list[int]:
message_ids: list[int] = []
for chunk in split_message_chunks(text, self.message_limit):
message = await self.tg_retry(lambda text=chunk: self.bot.send_message(text=text, **self.chat_kwargs()))
message_ids.append(int(message.message_id))
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
return message_ids
async def send_media_items(self, items: list[dict[str, Any]], caption: str | None = None) -> list[int]:
if not items:
return []
if len(items) == 1:
item = items[0]
if item["type"] == "photo":
message = await self.tg_retry(
lambda: self.bot.send_photo(photo=item["media"], caption=caption, **self.chat_kwargs())
)
else:
message = await self.tg_retry(
lambda: self.bot.send_video(video=item["media"], caption=caption, **self.chat_kwargs())
)
return [int(message.message_id)]
group = []
for idx, item in enumerate(items):
media_caption = caption if idx == 0 else None
if item["type"] == "photo":
group.append(InputMediaPhoto(media=item["media"], caption=media_caption))
else:
group.append(InputMediaVideo(media=item["media"], caption=media_caption))
messages = await self.tg_retry(lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()))
return [int(message.message_id) for message in messages]
async def send_post(self, post: dict[str, Any]) -> list[int]:
text = self.build_text(post)
media_rows = await self.load_media(int(post["id"]))
media_items = []
for item in media_rows:
media = self.media_value(item)
if media is None:
continue
media_type = str(item.get("media_type") or "")
if media_type == "photo":
media_items.append({"type": "photo", "media": media})
elif media_type == "video":
media_items.append({"type": "video", "media": media})
message_ids: list[int] = []
if not media_items:
return await self.send_text(text)
first_caption = text if len(text) <= self.caption_limit else self.overflow_caption[: self.caption_limit]
first = media_items[: self.media_group_max_items]
message_ids.extend(await self.send_media_items(first, first_caption))
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
rest = media_items[self.media_group_max_items :]
for start in range(0, len(rest), self.media_group_max_items):
chunk = rest[start : start + self.media_group_max_items]
message_ids.extend(await self.send_media_items(chunk))
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
if len(text) > self.caption_limit:
message_ids.extend(await self.send_text(text))
return message_ids
async def mark_published(self, post_id: int, message_ids: list[int]) -> None:
async with self.pool.acquire() as conn:
async with conn.transaction():
await conn.execute(
"""
UPDATE raw_posts
SET publication_status=$2,
editorial_status='published',
publication_error=NULL,
published_at=NOW(),
tg_publication_chat_id=$3,
tg_publication_thread_id=$4,
tg_publication_message_ids=$5,
tg_publication_url=$6,
updated_at=NOW()
WHERE id=$1
""",
post_id,
PUBLICATION_STATUS_PUBLISHED,
self.chat_id,
self.thread_id,
message_ids,
tg_message_url(int(self.chat_id), message_ids[0]) if message_ids else None,
)
await conn.execute(
"""
INSERT INTO post_publications(raw_post_id, platform, status, target_id, external_id, url, attachments_json, published_at, updated_at)
VALUES($1, 'tg', $2, $3, $4, $5, $6::jsonb, NOW(), NOW())
ON CONFLICT (raw_post_id, platform) DO UPDATE
SET status=EXCLUDED.status,
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()
""",
post_id,
PUBLICATION_STATUS_PUBLISHED,
str(self.chat_id),
",".join(str(mid) for mid in message_ids),
tg_message_url(int(self.chat_id), message_ids[0]) if message_ids else None,
json.dumps(message_ids, ensure_ascii=False),
)
async def mark_failed(self, post_id: int, error: str) -> None:
async with self.pool.acquire() as conn:
async with conn.transaction():
await conn.execute(
"""
UPDATE raw_posts
SET publication_status=$2,
editorial_status='publish_failed',
publication_error=$3,
updated_at=NOW()
WHERE id=$1
""",
post_id,
PUBLICATION_STATUS_FAILED,
error[:1000],
)
await conn.execute(
"""
INSERT INTO post_publications(raw_post_id, platform, status, target_id, error, updated_at)
VALUES($1, 'tg', $2, $3, $4, NOW())
ON CONFLICT (raw_post_id, platform) DO UPDATE
SET status=EXCLUDED.status,
target_id=EXCLUDED.target_id,
error=EXCLUDED.error,
updated_at=NOW()
""",
post_id,
PUBLICATION_STATUS_FAILED,
str(self.chat_id),
error[:1000],
)
async def process_slot(self, slot: dict[str, Any]) -> int:
count = int(slot["count"])
recent = await self.load_recent()
candidates = await self.load_candidates(max(50, count * 10))
posts = self.choose_posts(candidates, recent, count)
published = 0
for post in posts:
await self.heartbeat.beat(self.pool, status="publishing", meta={"post_id": int(post["id"]), "slot": slot["id"]}, force=True)
try:
message_ids = await self.send_post(post)
await self.mark_published(int(post["id"]), message_ids)
published += 1
logger.info("TG published raw_post={} messages={}", post["id"], message_ids)
except Exception as exc:
await self.mark_failed(int(post["id"]), str(exc))
logger.exception("TG publish failed raw_post={}: {}", post["id"], exc)
await self.pool.execute(
"""
UPDATE publication_runs
SET published_count=$2,
status=$3,
error=$4,
updated_at=NOW()
WHERE id=$1
""",
int(slot["run_id"]),
published,
"done" if published == count else "partial",
None if published == count else f"published {published} of {count}",
)
return published
async def run_once(self) -> bool:
await self.reload_settings()
enabled = await is_worker_enabled(self.pool, WORKER_TG_POSTER)
app_enabled = await fetch_bool_setting("tg_poster_enabled", False)
if not enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
slots = await self.due_slots()
if not slots:
await self.heartbeat.beat(self.pool, status="idle")
return False
total = 0
for slot in slots:
total += await self.process_slot(slot)
await self.heartbeat.beat(self.pool, status="idle", meta={"published": total}, force=True)
return bool(total)
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_TG_POSTER, os.getpid())
try:
while True:
await self.run_once()
await asyncio.sleep(max(10, await fetch_int_setting("tg_poster_interval_sec", 60)))
finally:
await self.close()
async def main() -> None:
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
worker = TelegramPoster()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
+326
View File
@@ -0,0 +1,326 @@
from __future__ import annotations
import asyncio
import json
import os
import re
import sys
from typing import Any
import aiohttp
from loguru import logger
from ..config import settings
from ..constants import PUBLICATION_STATUS_PUBLISHED, WORKER_TG_REACTOR
from ..db import fetch_bool_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
PLATFORM_TG = "tg"
REACTION_STATUS_PENDING = "pending"
REACTION_STATUS_IN_PROGRESS = "in_progress"
REACTION_STATUS_REACTED = "reacted"
REACTION_STATUS_FAILED = "failed"
def split_tokens(value: Any) -> list[str]:
if isinstance(value, list):
candidates = [str(item or "") for item in value]
else:
candidates = re.split(r"[\s,;]+", str(value or ""))
return [token.strip() for token in candidates if token.strip()]
def normalize_reactions(value: Any) -> list[str]:
if not isinstance(value, list):
value = []
reactions = []
for item in value:
emoji = str(item or "").strip()
if emoji:
reactions.append(emoji)
return reactions or ["👍"]
def parse_message_ids(row: dict[str, Any]) -> list[str]:
raw = row.get("attachments_json")
values: list[Any] = []
if isinstance(raw, list):
values = raw
elif isinstance(raw, str):
try:
parsed = json.loads(raw)
if isinstance(parsed, list):
values = parsed
except json.JSONDecodeError:
pass
if not values and row.get("external_id"):
values = str(row["external_id"]).split(",")
message_ids = []
for value in values:
text = str(value or "").strip()
if text:
message_ids.append(text)
return message_ids
class TelegramReactor:
def __init__(self) -> None:
self.pool = None
self.heartbeat = HeartbeatReporter(WORKER_TG_REACTOR, 30)
self.bot_tokens: list[str] = []
self.reactions: list[str] = ["👍"]
self.delay_sec = 60
self.interval_sec = 60
self.reaction_pause_sec = 2
self.react_all_messages = False
self.max_attempts = 3
self.retry_backoff_max_sec = 30
self.since = ""
async def init(self) -> None:
self.pool = await get_pool()
await self.reload_settings()
async def reload_settings(self) -> None:
tokens = split_tokens(await fetch_setting("tg_reactor_bot_tokens", ""))
if not tokens:
fallback = str(await fetch_setting("tg_poster_bot_token", "") or "").strip() or settings.tg_bot_token
tokens = split_tokens(fallback)
if not tokens:
raise RuntimeError("tg_reactor_bot_tokens, tg_poster_bot_token and TG_BOT_TOKEN are empty")
self.bot_tokens = tokens
self.reactions = normalize_reactions(await fetch_setting("tg_reactor_reactions_json", ["👍"]))
self.since = str(await fetch_setting("tg_reactor_since", "") or "").strip()
self.delay_sec = max(0, await fetch_int_setting("tg_reactor_delay_sec", 60))
self.interval_sec = max(10, await fetch_int_setting("tg_reactor_interval_sec", 60))
self.reaction_pause_sec = max(0, await fetch_int_setting("tg_reactor_reaction_pause_sec", 2))
self.react_all_messages = await fetch_bool_setting("tg_reactor_react_all_messages", False)
self.max_attempts = max(1, await fetch_int_setting("tg_reactor_max_attempts", 3))
self.retry_backoff_max_sec = max(1, await fetch_int_setting("tg_reactor_retry_backoff_max_sec", 30))
logger.info(
"TG reactor config: bots={} reactions={} since={} delay={}s pause={}s all_messages={}",
len(self.bot_tokens),
self.reactions,
self.since or "-",
self.delay_sec,
self.reaction_pause_sec,
self.react_all_messages,
)
async def load_publications(self, limit: int = 50) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT pp.*
FROM post_publications pp
WHERE pp.platform=$1
AND pp.status=$2
AND pp.published_at IS NOT NULL
AND pp.published_at <= NOW() - ($3::int * INTERVAL '1 second')
AND ($4::text = '' OR pp.published_at >= $4::timestamptz)
ORDER BY pp.published_at DESC, pp.id DESC
LIMIT $5::int
""",
PLATFORM_TG,
PUBLICATION_STATUS_PUBLISHED,
self.delay_sec,
self.since,
limit,
)
return [dict(row) for row in rows]
async def ensure_reaction_row(
self,
publication: dict[str, Any],
reactor_key: str,
reaction: str,
message_id: str,
) -> dict[str, Any] | None:
row = await self.pool.fetchrow(
"""
INSERT INTO post_reactions(
publication_id, raw_post_id, platform, reactor_key, reaction, target_id, message_id, status, updated_at
)
VALUES($1, $2, $3, $4, $5, $6, $7, $8, NOW())
ON CONFLICT (publication_id, reactor_key, message_id) DO NOTHING
RETURNING *
""",
int(publication["id"]),
int(publication["raw_post_id"]),
PLATFORM_TG,
reactor_key,
reaction,
str(publication["target_id"]),
message_id,
REACTION_STATUS_PENDING,
)
if row:
return dict(row)
row = await self.pool.fetchrow(
"""
SELECT *
FROM post_reactions
WHERE publication_id=$1 AND reactor_key=$2 AND message_id=$3
""",
int(publication["id"]),
reactor_key,
message_id,
)
if not row or row["status"] == REACTION_STATUS_REACTED or int(row["attempts"] or 0) >= self.max_attempts:
return None
return dict(row)
async def claim_reaction(self, reaction_id: int) -> dict[str, Any] | None:
row = await self.pool.fetchrow(
"""
UPDATE post_reactions
SET status=$2,
attempts=attempts + 1,
error=NULL,
updated_at=NOW()
WHERE id=$1
AND status IN ($3, $4)
AND attempts < $5
RETURNING *
""",
reaction_id,
REACTION_STATUS_IN_PROGRESS,
REACTION_STATUS_PENDING,
REACTION_STATUS_FAILED,
self.max_attempts,
)
return dict(row) if row else None
async def set_reaction(self, token: str, chat_id: str, message_id: str, reaction: str) -> None:
url = f"https://api.telegram.org/bot{token}/setMessageReaction"
payload = {
"chat_id": int(chat_id),
"message_id": int(message_id),
"reaction": [{"type": "emoji", "emoji": reaction}],
"is_big": False,
}
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
async with session.post(url, json=payload) as resp:
data = await resp.json(content_type=None)
if not data.get("ok"):
description = data.get("description") or data
retry_after = ((data.get("parameters") or {}).get("retry_after")) if isinstance(data, dict) else None
if retry_after:
raise RuntimeError(f"Telegram retry_after={retry_after}: {description}")
raise RuntimeError(str(description))
async def mark_reacted(self, reaction_id: int) -> None:
await self.pool.execute(
"""
UPDATE post_reactions
SET status=$2,
error=NULL,
reacted_at=NOW(),
updated_at=NOW()
WHERE id=$1
""",
reaction_id,
REACTION_STATUS_REACTED,
)
async def mark_failed(self, reaction_id: int, error: str) -> None:
await self.pool.execute(
"""
UPDATE post_reactions
SET status=$2,
error=$3,
updated_at=NOW()
WHERE id=$1
""",
reaction_id,
REACTION_STATUS_FAILED,
error[:1000],
)
async def process_publication(self, publication: dict[str, Any]) -> int:
message_ids = parse_message_ids(publication)
if not message_ids:
return 0
if not self.react_all_messages:
message_ids = message_ids[:1]
done = 0
for message_id in message_ids:
for idx, token in enumerate(self.bot_tokens):
reaction = self.reactions[idx % len(self.reactions)]
reactor_key = f"bot-{idx + 1}:{reaction}"
row = await self.ensure_reaction_row(publication, reactor_key, reaction, message_id)
if not row:
continue
claimed = await self.claim_reaction(int(row["id"]))
if not claimed:
continue
try:
await self.set_reaction(token, str(publication["target_id"]), message_id, reaction)
await self.mark_reacted(int(claimed["id"]))
done += 1
logger.info(
"TG reacted raw_post={} publication={} message={} reactor={}",
publication["raw_post_id"],
publication["id"],
message_id,
reactor_key,
)
if self.reaction_pause_sec:
await asyncio.sleep(self.reaction_pause_sec)
except Exception as exc:
await self.mark_failed(int(claimed["id"]), str(exc))
logger.warning(
"TG reaction failed raw_post={} publication={} message={} reactor={}: {}",
publication["raw_post_id"],
publication["id"],
message_id,
reactor_key,
exc,
)
await asyncio.sleep(min(2 ** int(claimed["attempts"]), self.retry_backoff_max_sec))
return done
async def run_once(self) -> bool:
await self.reload_settings()
enabled = await is_worker_enabled(self.pool, WORKER_TG_REACTOR)
app_enabled = await fetch_bool_setting("tg_reactor_enabled", False)
if not enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
publications = await self.load_publications()
if not publications:
await self.heartbeat.beat(self.pool, status="idle")
return False
total = 0
for publication in publications:
await self.heartbeat.beat(
self.pool,
status="reacting",
meta={"publication_id": int(publication["id"]), "raw_post_id": int(publication["raw_post_id"])},
force=True,
)
total += await self.process_publication(publication)
await self.heartbeat.beat(self.pool, status="idle", meta={"reactions": total}, force=True)
return bool(total)
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_TG_REACTOR, os.getpid())
while True:
try:
await self.run_once()
except Exception as exc:
logger.exception("TG reactor loop error: {}", exc)
await asyncio.sleep(self.interval_sec)
async def main() -> None:
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
worker = TelegramReactor()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
+442
View File
@@ -0,0 +1,442 @@
from __future__ import annotations
import asyncio
import json
import os
import re
import sys
from datetime import datetime, time, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo
import aiohttp
from loguru import logger
from ..config import settings
from ..constants import (
PUBLICATION_STATUS_FAILED,
PUBLICATION_STATUS_PENDING,
PUBLICATION_STATUS_PUBLISHED,
WORKER_VK_POSTER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import build_publication_text
from ..vk_api import VKAPIClient, post_vk_url
LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
VK_MESSAGE_LIMIT = 16384
VK_MAX_ATTACHMENTS = 10
def valid_time(value: str) -> bool:
return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))
def normalize_schedule(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list):
value = []
rows: list[dict[str, Any]] = []
for idx, item in enumerate(value):
if not isinstance(item, dict):
continue
time_value = str(item.get("time") or "").strip()
if not valid_time(time_value):
continue
count = max(0, min(20, int(item.get("count") or 0)))
if count < 1:
continue
rows.append(
{
"id": str(item.get("id") or f"slot-{idx + 1}").strip()[:80],
"time": time_value,
"count": count,
"enabled": bool(item.get("enabled", True)),
}
)
return sorted(rows, key=lambda row: row["time"])
class VKPoster:
def __init__(self) -> None:
self.pool = None
self.heartbeat = HeartbeatReporter(WORKER_VK_POSTER, 30)
self.token = ""
self.owner_id = 0
self.from_group = True
self.schedule: list[dict[str, Any]] = []
self.interval_sec = 60
self.send_delay_sec = 1.0
self.recent_window = 20
self.category_repeat_penalty = 3.0
self.source_repeat_penalty = 4.0
self.dry_run = False
async def init(self) -> None:
self.pool = await get_pool()
def token_expires_soon(self, value: Any) -> bool:
raw = str(value or "").strip()
if not raw:
return False
try:
expires_at = datetime.fromisoformat(raw)
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
except ValueError:
return False
return expires_at <= datetime.now(timezone.utc) + timedelta(minutes=15)
async def save_setting(self, key: str, value: Any) -> None:
await self.pool.execute(
"""
UPDATE app_settings
SET value_json=$2::jsonb,
updated_at=NOW()
WHERE key=$1
""",
key,
json.dumps(value, ensure_ascii=False),
)
async def refresh_access_token_if_needed(self) -> None:
expires_at = await fetch_setting("vk_poster_token_expires_at", "")
if not self.token_expires_soon(expires_at):
return
refresh_token = str(await fetch_setting("vk_poster_refresh_token", "") or "").strip()
client_id = str(await fetch_setting("vk_poster_app_id", "") or "").strip()
client_secret = str(await fetch_setting("vk_poster_client_secret", "") or "").strip()
device_id = str(await fetch_setting("vk_poster_token_device_id", "") or "").strip()
if not refresh_token or not client_id:
return
data = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"device_id": device_id,
"state": "vk-poster-refresh",
}
if client_secret:
data["client_secret"] = client_secret
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
async with session.post("https://id.vk.ru/oauth2/auth", data=data) as resp:
payload = await resp.json(content_type=None)
if payload.get("error"):
raise RuntimeError(f"VK token refresh failed: {payload.get('error')}: {payload.get('error_description')}")
access_token = str(payload.get("access_token") or "").strip()
if not access_token:
raise RuntimeError(f"VK token refresh did not return access_token: {payload}")
new_refresh = str(payload.get("refresh_token") or "").strip() or refresh_token
new_device_id = str(payload.get("device_id") or "").strip() or device_id
expires_in = payload.get("expires_in")
expires_at_value = ""
try:
expires_at_value = (datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))).isoformat()
except Exception:
pass
await self.save_setting("vk_poster_access_token", access_token)
await self.save_setting("vk_poster_refresh_token", new_refresh)
await self.save_setting("vk_poster_token_device_id", new_device_id)
await self.save_setting("vk_poster_token_expires_at", expires_at_value)
self.token = access_token
logger.info("VK access token refreshed; expires_at={}", expires_at_value)
async def reload_settings(self) -> None:
self.token = (
str(await fetch_setting("vk_poster_access_token", "") or "").strip()
or settings.vk_group_access_token
or settings.vk_access_token
)
await self.refresh_access_token_if_needed()
self.owner_id = int(await fetch_int_setting("vk_poster_owner_id", settings.vk_storage_owner_id if settings.vk_storage_group_id else 0))
self.from_group = await fetch_bool_setting("vk_poster_from_group", True)
self.schedule = normalize_schedule(await fetch_setting("vk_poster_schedule_json", []))
self.interval_sec = max(10, await fetch_int_setting("vk_poster_interval_sec", 60))
self.send_delay_sec = max(0.0, await fetch_float_setting("vk_poster_send_delay_sec", 1.0))
self.recent_window = max(1, await fetch_int_setting("vk_poster_recent_window", 20))
self.category_repeat_penalty = max(0.0, await fetch_float_setting("vk_poster_category_repeat_penalty", 3.0))
self.source_repeat_penalty = max(0.0, await fetch_float_setting("vk_poster_source_repeat_penalty", 4.0))
self.dry_run = await fetch_bool_setting("vk_poster_dry_run", False)
if not self.token:
raise RuntimeError("vk_poster_access_token, VK_GROUP_ACCESS_TOKEN and VK_ACCESS_TOKEN are empty")
if not self.owner_id:
raise RuntimeError("vk_poster_owner_id is empty")
logger.info("VK poster config: owner={} schedule={} dry_run={}", self.owner_id, self.schedule, self.dry_run)
async def due_slots(self) -> list[dict[str, Any]]:
now = datetime.now(LOCAL_TZ)
today = now.date()
due: list[dict[str, Any]] = []
for row in self.schedule:
if not row.get("enabled"):
continue
hour, minute = [int(part) for part in str(row["time"]).split(":", 1)]
slot_start = datetime.combine(today, time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
slot_end = slot_start + timedelta(minutes=10)
if not (slot_start <= now < slot_end):
continue
inserted = await self.pool.fetchrow(
"""
INSERT INTO publication_runs(poster, schedule_id, scheduled_for, scheduled_time, planned_count, status)
VALUES($1, $2, $3, $4, $5, 'started')
ON CONFLICT (poster, schedule_id, scheduled_for) DO NOTHING
RETURNING id
""",
"vk",
row["id"],
today,
row["time"],
int(row["count"]),
)
if inserted:
due.append({**row, "run_id": int(inserted["id"]), "date": today})
return due
async def load_recent(self) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT rp.final_category_tag, rp.rewrite_category_tag, rp.final_source_tag, rp.rewrite_source_tag
FROM post_publications pp
JOIN raw_posts rp ON rp.id=pp.raw_post_id
WHERE pp.platform='vk'
AND pp.status=$1
AND pp.target_id=$2
ORDER BY pp.published_at DESC NULLS LAST, pp.id DESC
LIMIT $3::int
""",
PUBLICATION_STATUS_PUBLISHED,
str(self.owner_id),
self.recent_window,
)
return [dict(row) for row in rows]
async def load_candidates(self, limit: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT rp.*, s.name AS source_name, s.tag AS source_tag
FROM raw_posts rp
JOIN sources s ON s.id=rp.source_id
LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='vk'
WHERE rp.status='storage_ready'
AND rp.rewrite_status='ready'
AND COALESCE(rp.editorial_status, 'review') IN ('accepted', 'published', 'publish_failed')
AND COALESCE(pp.status, $1) = $1
AND EXISTS (
SELECT 1
FROM raw_post_media rpm
WHERE rpm.raw_post_id=rp.id
AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
AND COALESCE(rpm.editor_added, FALSE)=FALSE
AND (
(rpm.media_type='photo' AND rpm.original_attachment_id LIKE 'photo%')
OR (
rpm.media_type='video'
AND rpm.original_attachment_id LIKE 'video%'
AND COALESCE(rpm.status, '') <> 'link_only'
)
)
)
ORDER BY COALESCE(rp.reviewed_at, rp.edited_at, rp.rewritten_at, rp.created_at) ASC,
rp.id ASC
LIMIT $2::int
""",
PUBLICATION_STATUS_PENDING,
max(limit, 50),
)
return [dict(row) for row in rows]
def choose_posts(self, candidates: list[dict[str, Any]], recent: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
selected: list[dict[str, Any]] = []
recent_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in recent]
recent_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or "") for row in recent]
pool = list(candidates)
while pool and len(selected) < count:
best_idx = 0
selected_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in selected]
selected_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or row.get("source_tag") or "") for row in selected]
previous_category = selected_categories[-1] if selected_categories else (recent_categories[0] if recent_categories else "")
previous_source = selected_sources[-1] if selected_sources else (recent_sources[0] if recent_sources else "")
best_rank: tuple[int, float, int] | None = None
for idx, post in enumerate(pool):
category = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or "")
source = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
immediate_repeats = int(bool(category) and category == previous_category) + int(bool(source) and source == previous_source)
repeat_penalty = (
(recent_categories.count(category) + selected_categories.count(category)) * self.category_repeat_penalty
+ (recent_sources.count(source) + selected_sources.count(source)) * self.source_repeat_penalty
)
rank = (immediate_repeats, repeat_penalty, idx)
if best_rank is None or rank < best_rank:
best_idx = idx
best_rank = rank
selected.append(pool.pop(best_idx))
return selected
async def load_media(self, raw_post_id: int) -> list[dict[str, Any]]:
rows = await self.pool.fetch(
"""
SELECT *
FROM raw_post_media
WHERE raw_post_id=$1
AND COALESCE(editor_hidden, FALSE)=FALSE
AND media_type IN ('photo', 'video')
ORDER BY sort_order ASC, id ASC
LIMIT $2::int
""",
raw_post_id,
VK_MAX_ATTACHMENTS,
)
return [dict(row) for row in rows]
def build_text(self, post: dict[str, Any]) -> str:
category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag)[:VK_MESSAGE_LIMIT]
async def media_attachment(self, client: VKAPIClient, item: dict[str, Any]) -> str | None:
attachment_id = str(item.get("original_attachment_id") or "").strip()
media_type = str(item.get("media_type") or "").strip()
if media_type == "video":
if str(item.get("status") or "").strip() == "link_only":
return None
return attachment_id if attachment_id.startswith("video") else None
if media_type != "photo":
return None
if attachment_id.startswith("photo") and not item.get("editor_added"):
return attachment_id
return None
async def send_post(self, post: dict[str, Any]) -> tuple[int, str, list[str]]:
text = self.build_text(post)
media_rows = await self.load_media(int(post["id"]))
async with VKAPIClient(
token=self.token,
rps=2,
timeout_total_sec=max(30, await fetch_int_setting("vk_poster_timeout_sec", 90)),
retry_attempts=max(1, await fetch_int_setting("vk_poster_max_attempts", 3)),
retry_min_delay_sec=2,
retry_max_delay_sec=max(2, await fetch_int_setting("vk_poster_retry_backoff_max_sec", 30)),
) as client:
attachments: list[str] = []
for item in media_rows:
try:
attachment = await self.media_attachment(client, item)
except Exception as exc:
logger.warning("VK media skipped raw_post={} media_id={}: {}", post["id"], item.get("id"), exc)
continue
if attachment:
attachments.append(attachment)
if len(attachments) >= VK_MAX_ATTACHMENTS:
break
if self.dry_run:
return 0, "", attachments
post_id = await client.create_wall_post(self.owner_id, text, attachments, from_group=self.from_group)
return post_id, post_vk_url(self.owner_id, post_id), attachments
async def upsert_publication(
self,
post_id: int,
status: str,
external_id: str | None = None,
url: str | None = None,
error: str | None = None,
attachments: list[str] | None = None,
) -> None:
await self.pool.execute(
"""
INSERT INTO post_publications(
raw_post_id, platform, status, target_id, external_id, url, error, attachments_json, published_at, updated_at
)
VALUES($1, 'vk', $2, $3, $4, $5, $6, $7::jsonb, CASE WHEN $2=$8 THEN NOW() ELSE NULL END, NOW())
ON CONFLICT (raw_post_id, platform) DO UPDATE
SET status=EXCLUDED.status,
target_id=EXCLUDED.target_id,
external_id=EXCLUDED.external_id,
url=EXCLUDED.url,
error=EXCLUDED.error,
attachments_json=EXCLUDED.attachments_json,
published_at=CASE WHEN EXCLUDED.status=$8 THEN COALESCE(EXCLUDED.published_at, NOW()) ELSE post_publications.published_at END,
updated_at=NOW()
""",
post_id,
status,
str(self.owner_id),
external_id,
url,
error[:1000] if error else None,
json.dumps(attachments or [], ensure_ascii=False),
PUBLICATION_STATUS_PUBLISHED,
)
async def process_slot(self, slot: dict[str, Any]) -> int:
count = int(slot["count"])
recent = await self.load_recent()
candidates = await self.load_candidates(max(50, count * 10))
posts = self.choose_posts(candidates, recent, count)
published = 0
for post in posts:
await self.heartbeat.beat(self.pool, status="publishing", meta={"post_id": int(post["id"]), "slot": slot["id"]}, force=True)
try:
vk_post_id, url, attachments = await self.send_post(post)
await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_PUBLISHED, str(vk_post_id), url, attachments=attachments)
published += 1
logger.info("VK published raw_post={} post_id={} attachments={}", post["id"], vk_post_id, len(attachments))
except Exception as exc:
await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_FAILED, error=str(exc))
logger.exception("VK publish failed raw_post={}: {}", post["id"], exc)
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
await self.pool.execute(
"""
UPDATE publication_runs
SET published_count=$2,
status=$3,
error=$4,
updated_at=NOW()
WHERE id=$1
""",
int(slot["run_id"]),
published,
"done" if published == count else "partial",
None if published == count else f"published {published} of {count}",
)
return published
async def run_once(self) -> bool:
await self.reload_settings()
enabled = await is_worker_enabled(self.pool, WORKER_VK_POSTER)
app_enabled = await fetch_bool_setting("vk_poster_enabled", False)
if not enabled or not app_enabled:
await self.heartbeat.beat(self.pool, status="disabled", force=True)
return False
slots = await self.due_slots()
if not slots:
await self.heartbeat.beat(self.pool, status="idle")
return False
total = 0
for slot in slots:
total += await self.process_slot(slot)
await self.heartbeat.beat(self.pool, status="idle", meta={"published": total}, force=True)
return bool(total)
async def run_loop(self) -> None:
await self.init()
logger.info("{} started pid={}", WORKER_VK_POSTER, os.getpid())
while True:
try:
await self.run_once()
except Exception as exc:
logger.exception("VK poster loop error: {}", exc)
await asyncio.sleep(self.interval_sec)
async def main() -> None:
logger.remove()
logger.add(sys.stdout, level=settings.log_level)
worker = VKPoster()
await worker.run_loop()
if __name__ == "__main__":
asyncio.run(main())
@@ -91,6 +91,7 @@ class TelegramStorageUploader:
self.text_overflow_marker = "Продолжение следующим сообщением."
self.download_timeout_sec = 45
self.video_max_size_mb = 49
self.video_max_height = 720
self.video_max_duration_sec = 300
self.yt_dlp_timeout_sec = 300
self.max_media_attempts = 3
@@ -126,6 +127,7 @@ class TelegramStorageUploader:
self.text_overflow_marker = str(await fetch_setting("telegram_text_overflow_marker", self.text_overflow_marker))
self.download_timeout_sec = max(5, await fetch_int_setting("uploader_download_timeout_sec", 45))
self.video_max_size_mb = max(1, await fetch_int_setting("video_max_size_mb", 49))
self.video_max_height = max(144, await fetch_int_setting("video_max_height", 720))
self.video_max_duration_sec = max(1, await fetch_int_setting("video_max_duration_sec", 300))
self.yt_dlp_timeout_sec = max(30, await fetch_int_setting("uploader_yt_dlp_timeout_sec", 300))
self.max_media_attempts = max(1, await fetch_int_setting("uploader_max_media_attempts", 3))
@@ -192,6 +194,64 @@ class TelegramStorageUploader:
)
return [dict(row) for row in rows]
def media_blocking_error(self, media: list[dict], prepared_media_ids: set[int] | None = None) -> str | None:
prepared_media_ids = prepared_media_ids or set()
visible_media = [
item
for item in media
if str(item.get("media_type") or "") in ("photo", "video")
and not bool(item.get("editor_hidden"))
]
if not visible_media:
return "no publishable media"
pending = [
item
for item in visible_media
if str(item.get("status") or "") == "pending"
and int(item["id"]) not in prepared_media_ids
]
if pending:
details = ", ".join(
f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'pending'}"
for item in pending[:5]
)
return f"media still pending: {details}"
failed = [
item
for item in visible_media
if str(item.get("status") or "") == MEDIA_STATUS_FAILED
]
if failed:
details = ", ".join(
f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'failed'}"
for item in failed[:5]
)
return f"media failed: {details}"
link_only = [
item
for item in visible_media
if str(item.get("status") or "") == MEDIA_STATUS_LINK_ONLY
]
if link_only:
details = ", ".join(
f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'link only'}"
for item in link_only[:5]
)
return f"media link only: {details}"
missing_upload = [
item
for item in visible_media
if not (item.get("tg_file_id") or item.get("storage_attachment_id"))
and int(item["id"]) not in prepared_media_ids
]
if missing_upload:
details = ", ".join(
f"{item.get('media_type')}#{item.get('id')}: missing tg_file_id"
for item in missing_upload[:5]
)
return f"media not uploaded: {details}"
return None
async def download_bytes(self, session: aiohttp.ClientSession, url: str) -> bytes | None:
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.download_timeout_sec)) as response:
@@ -271,7 +331,13 @@ class TelegramStorageUploader:
output_path,
"--no-playlist",
"-f",
f"best[filesize<{max_size}]/bestvideo[filesize<{max_size}]+bestaudio/best",
(
f"best[height<={self.video_max_height}][filesize<{max_size}]"
f"/best[height<={self.video_max_height}]"
f"/bestvideo[height<={self.video_max_height}][filesize<{max_size}]+bestaudio/best"
f"/bestvideo[height<={self.video_max_height}]+bestaudio/best"
f"/best[filesize<{max_size}]"
),
"--quiet",
"--no-warnings",
]
@@ -498,6 +564,25 @@ class TelegramStorageUploader:
media = await self.load_media(raw_post_id)
prepared = await self.prepare_media(raw_post_id, media)
media = await self.load_media(raw_post_id)
prepared_media_ids = {int(item["media_id"]) for item in prepared}
blocking_error = self.media_blocking_error(media, prepared_media_ids)
if blocking_error:
for item in prepared:
tmp_path = item.get("tmp_path")
if tmp_path:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
if not blocking_error.startswith("media still pending:"):
await self.mark_post_failed(raw_post_id, blocking_error)
await ack_done(self.pool, job_id)
logger.warning("Telegram storage blocked raw_post={}: {}", raw_post_id, blocking_error)
return
await ack_retry(self.pool, job, blocking_error, delay_sec=120)
logger.warning("Telegram storage retry raw_post={}: {}", raw_post_id, blocking_error)
return
try:
message_ids, meta_message_id = await self.send_storage_post(post, prepared, media)
finally: