exostring/new_vk_parser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch prompt test to writer

Browse Source
This commit is contained in:
Your Name
2026-06-16 01:02:08 +05:00
parent d306ca42a1
commit 85a76652fe
2 changed files with 48 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vk_parser_app/admin.py
+39 -24
View File
@@ -922,6 +922,7 @@ async def prompt_test_posts(selected_post_id: int | None = None) -> tuple[list[d
"""
SELECT rp.id, rp.raw_text, rp.original_url, rp.posted_at, rp.created_at, rp.qualification_status,
rp.qualification_score, s.name AS source_name, s.tag AS source_tag,
rp.qualification_reason,
(SELECT COUNT(*) FROM raw_post_media m WHERE m.raw_post_id=rp.id) AS media_count,
(SELECT ARRAY_AGG(DISTINCT m.media_type ORDER BY m.media_type)
FROM raw_post_media m WHERE m.raw_post_id=rp.id) AS media_types
@@ -946,6 +947,7 @@ async def prompt_test_posts(selected_post_id: int | None = None) -> tuple[list[d
"""
SELECT rp.id, rp.raw_text, rp.original_url, rp.posted_at, rp.created_at, rp.qualification_status,
rp.qualification_score, s.name AS source_name, s.tag AS source_tag,
rp.qualification_reason,
(SELECT COUNT(*) FROM raw_post_media m WHERE m.raw_post_id=rp.id) AS media_count,
(SELECT ARRAY_AGG(DISTINCT m.media_type ORDER BY m.media_type)
FROM raw_post_media m WHERE m.raw_post_id=rp.id) AS media_types
@@ -968,17 +970,24 @@ async def prompt_test_posts(selected_post_id: int | None = None) -> tuple[list[d
return posts, selected
def prompt_test_payload(post: dict[str, Any], max_text_chars: int) -> list[dict[str, Any]]:
return [
{
"id": int(post["id"]),
"source": post.get("source_name") or "",
"original_url": post.get("original_url") or "",
"media_count": int(post.get("media_count") or 0),
"media_types": list(post.get("media_types") or []),
"text": str(post.get("raw_text") or "")[:max_text_chars],
}
]
def prompt_test_payload(post: dict[str, Any], max_text_chars: int) -> dict[str, Any]:
return {
"task": "rewrite_accepted_posts",
"categories": [],
"posts": [
{
"id": int(post["id"]),
"producer_name": post.get("source_name") or "",
"producer_tag": normalize_hash_tag(post.get("source_tag") or post.get("source_name") or "", "source"),
"original_url": post.get("original_url") or "",
"qualification_score": post.get("qualification_score"),
"qualification_reason": post.get("qualification_reason") or "",
"media_count": int(post.get("media_count") or 0),
"media_types": list(post.get("media_types") or []),
"text": str(post.get("raw_text") or "")[:max_text_chars],
}
],
}
def setting_value(settings_rows: list[dict], key: str, default: Any = None) -> Any:
@@ -2268,11 +2277,14 @@ async def prompt_test(request: Request, post_id: int | None = None):
if not user:
return redirect("/login")
posts, selected = await prompt_test_posts(post_id)
max_text_chars = max(100, int(await fetch_setting("ai_qualifier_max_text_chars", 2000) or 2000))
prompt = str(await fetch_setting("ai_qualifier_prompt", "") or "")
provider = str(await fetch_setting("ai_qualifier_provider", "openrouter") or "openrouter")
model = str(await fetch_setting("ai_qualifier_model", "") or "")
categories = await writer_categories()
max_text_chars = max(200, int(await fetch_setting("ai_writer_max_text_chars", 3500) or 3500))
prompt = str(await fetch_setting("ai_writer_prompt", "") or "")
provider = str(await fetch_setting("ai_writer_provider", "anthropic") or "anthropic")
model = str(await fetch_setting("ai_writer_model", "") or "")
payload = prompt_test_payload(selected, max_text_chars) if selected else []
if payload:
payload["categories"] = categories
return templates.TemplateResponse(
"prompt_test.html",
base_context(
@@ -2303,21 +2315,24 @@ async def prompt_test_run(
return redirect("/login")
require_csrf(user, csrf_token)
posts, selected = await prompt_test_posts(post_id)
max_text_chars = max(100, int(await fetch_setting("ai_qualifier_max_text_chars", 2000) or 2000))
provider = str(await fetch_setting("ai_qualifier_provider", "openrouter") or "openrouter")
model = str(await fetch_setting("ai_qualifier_model", "") or "").strip()
api_key = str(await fetch_setting("ai_qualifier_api_key", "") or "").strip()
api_base = str(await fetch_setting("ai_qualifier_api_base", "") or "").strip()
temperature = max(0.0, float(await fetch_setting("ai_qualifier_temperature", 0.0) or 0.0))
timeout = max(10, int(await fetch_setting("ai_qualifier_timeout_sec", 120) or 120))
categories = await writer_categories()
max_text_chars = max(200, int(await fetch_setting("ai_writer_max_text_chars", 3500) or 3500))
provider = str(await fetch_setting("ai_writer_provider", "anthropic") or "anthropic")
model = str(await fetch_setting("ai_writer_model", "") or "").strip()
api_key = str(await fetch_setting("ai_writer_api_key", "") or "").strip()
api_base = str(await fetch_setting("ai_writer_api_base", "") or "").strip()
temperature = max(0.0, float(await fetch_setting("ai_writer_temperature", 0.4) or 0.4))
timeout = max(10, int(await fetch_setting("ai_writer_timeout_sec", 180) or 180))
normalized_prompt = prompt.replace("\\r\\n", "\n").replace("\\n", "\n").strip()
payload = prompt_test_payload(selected, max_text_chars) if selected else []
if payload:
payload["categories"] = categories
result = None
error = None
if not selected:
error = "Пост не найден."
elif not model or not api_key:
error = "Не заполнены модель или API-ключ квалификатора в настройках."
error = "Не заполнены модель или API-ключ райтера в настройках."
elif not normalized_prompt:
error = "Промпт пустой."
else:
@@ -2355,7 +2370,7 @@ async def prompt_test_run(
"usage": response_usage(response),
"model": kwargs["model"],
}
await audit(user["id"], "prompt_test.run", "raw_post", int(selected["id"]), {"provider": provider, "model": model})
await audit(user["id"], "writer_prompt_test.run", "raw_post", int(selected["id"]), {"provider": provider, "model": model})
except Exception as exc:
logger.exception("Prompt test failed")
error = str(exc)