# retoor <retoor@molodetz.nl>
import asyncio
import base64
import json
import logging
from datetime import datetime, timezone
from io import BytesIO
from devplacepy import stealth
from devplacepy.attachments import link_attachments, store_attachment
from devplacepy.awards.images import enforce_rgba_png, resize_award_png
from devplacepy.config import (
AWARD_GENERATION_TIMEOUT_SECONDS,
AWARD_IMAGE_MODEL_DEFAULT,
AWARD_IMAGE_PROMPT_DEFAULT,
AWARD_IMAGE_SIZE_DEFAULT,
INTERNAL_BASE_URL,
)
from devplacepy.database import add_award_usage, get_award_usage, get_setting, get_table
from devplacepy.database.awards import recompute_user_award_stats
from devplacepy.services.base import ConfigField
from devplacepy.services.correction import new_usage_totals
from devplacepy.services.jobs.base import JobService
from devplacepy.services.openai_gateway.usage import accumulate_usage, usage_metric_cards
from devplacepy.utils import create_notification
logger = logging.getLogger(__name__)
class AwardService(JobService):
kind = "award"
title = "Awards"
description = (
"Generates award images off the request path, stores PNG attachments, and "
"notifies receivers when ready."
)
def __init__(self):
super().__init__(name="award", interval_seconds=2)
self.config_fields = list(self.config_fields) + [
ConfigField(
"award_give_cooldown_hours",
"Give cooldown (hours)",
type="int",
default=24,
minimum=1,
maximum=168,
group="Awards",
),
ConfigField(
"award_receive_cooldown_hours",
"Receive cooldown (hours)",
type="int",
default=24,
minimum=1,
maximum=168,
group="Awards",
),
ConfigField(
"award_display_hours",
"Prominent display (hours)",
type="int",
default=24,
minimum=1,
maximum=168,
group="Awards",
),
ConfigField(
"award_image_model",
"Image model",
type="str",
default=AWARD_IMAGE_MODEL_DEFAULT,
group="Awards",
),
ConfigField(
"award_image_prompt",
"Image prompt",
type="text",
default=AWARD_IMAGE_PROMPT_DEFAULT,
group="Awards",
),
]
async def process(self, job: dict) -> dict:
totals = new_usage_totals()
try:
result = await asyncio.to_thread(self._generate, job, totals)
except Exception as exc:
self._audit_failed(job, str(exc))
raise
if totals["calls"]:
add_award_usage(totals)
return result
def _generate(self, job: dict, totals: dict) -> dict:
from devplacepy.services.audit import record as audit
payload = job.get("payload") or {}
award_uid = str(payload.get("award_uid", ""))
giver_uid = str(payload.get("giver_uid", ""))
receiver_uid = str(payload.get("receiver_uid", ""))
description = str(payload.get("description", ""))
api_key = str(payload.get("api_key", ""))
awards = get_table("awards")
row = awards.find_one(uid=award_uid)
if not row or row.get("deleted_at") or row.get("generated_at"):
return {"award_uid": award_uid, "skipped": True}
source_png = self._fetch_image(api_key, description, totals)
source_png = enforce_rgba_png(source_png)
uid512 = store_attachment(
resize_award_png(source_png, 512), "award-512.png", receiver_uid
)["uid"]
uid256 = store_attachment(
resize_award_png(source_png, 256), "award-256.png", giver_uid
)["uid"]
uid64 = store_attachment(
resize_award_png(source_png, 64), "award-64.png", giver_uid
)["uid"]
link_attachments([uid512, uid256, uid64], "award", award_uid)
now = datetime.now(timezone.utc).isoformat()
awards.update(
{
"uid": award_uid,
"attachment_uid_512": uid512,
"attachment_uid_256": uid256,
"attachment_uid_64": uid64,
"generated_at": now,
},
["uid"],
)
recompute_user_award_stats(receiver_uid)
users = get_table("users")
giver = users.find_one(uid=giver_uid) or {}
receiver = users.find_one(uid=receiver_uid) or {}
slug = row.get("slug", "")
create_notification(
receiver_uid,
"award",
f"@{giver.get('username', 'someone')} gave you an award",
giver_uid,
f"/profile/{receiver.get('username', '')}?tab=awards#award-{slug}",
)
audit.record_system(
"award.complete",
actor_kind="service",
target_type="award",
target_uid=award_uid,
target_label=description[:80],
summary=f"Award ready for {receiver.get('username', receiver_uid)}",
links=[
audit.target("award", award_uid, slug),
audit.target("user", receiver_uid, receiver.get("username")),
audit.target("user", giver_uid, giver.get("username")),
audit.job(job.get("uid", "")),
],
)
return {"award_uid": award_uid, "slug": slug}
def _fetch_image(self, api_key: str, description: str, totals: dict) -> bytes:
prompt = get_setting("award_image_prompt", AWARD_IMAGE_PROMPT_DEFAULT)
model = get_setting("award_image_model", AWARD_IMAGE_MODEL_DEFAULT)
size = AWARD_IMAGE_SIZE_DEFAULT
body = {
"model": model,
"prompt": f"{prompt}\n\n{description}".strip(),
"size": size,
"background": "transparent",
"output_format": "png",
}
url = f"{INTERNAL_BASE_URL}/openai/v1/images/generations"
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
with stealth.stealth_sync_client(timeout=AWARD_GENERATION_TIMEOUT_SECONDS) as client:
response = client.post(url, json=body, headers=headers)
accumulate_usage(totals, response)
response.raise_for_status()
data = response.json()
items = data.get("data") or []
if not items:
raise RuntimeError("image API returned no data")
encoded = items[0].get("b64_json") or ""
if not encoded:
raise RuntimeError("image API returned no b64_json")
return base64.b64decode(encoded)
def _audit_failed(self, job: dict, error: str) -> None:
from devplacepy.services.audit import record as audit
payload = job.get("payload") or {}
award_uid = str(payload.get("award_uid", ""))
audit.record_system(
"award.failed",
actor_kind="service",
target_type="award",
target_uid=award_uid,
summary=f"Award generation failed: {error[:80]}",
metadata={"error": error[:500], "award_uid": award_uid},
result="failure",
links=[audit.job(job.get("uid", "")), audit.target("award", award_uid)],
)
def collect_metrics(self) -> dict:
base = super().collect_metrics()
base["stats"] = base.get("stats", []) + usage_metric_cards(get_award_usage())
return base
Reference in New Issue View Git Blame Copy Permalink