| Server IP : 3.147.158.171 / Your IP : 216.73.216.216 Web Server : Apache/2.4.67 (Amazon Linux) OpenSSL/3.5.5 System : Linux ip-172-31-2-178.us-east-2.compute.internal 6.1.172-216.329.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Wed May 20 06:31:34 UTC 2026 x86_64 User : ec2-user ( 1000) PHP Version : 8.4.21 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : OFF Directory : /tsai/repo/api/app/services/ |
Upload File : |
"""iCalendar (.ics) generation for character article posting schedules.
Pure logic — no I/O. Converts Character.articleSchedule into RFC 5545 VEVENTs
with RRULE recurrence so subscribed calendars stay current automatically.
Only schedules with day constraints (daysOfWeek or daysOfMonth) are emitted;
characters without a schedule are skipped.
"""
from datetime import datetime, date, timedelta
from typing import List, Optional, Tuple
from ..appTypes import Character, PostingRecord, SiteConfig
_BYDAY = ['SU', 'MO', 'TU', 'WE', 'TH', 'FR', 'SA']
def _escape_text(value: str) -> str:
if not value:
return ''
return (value.replace('\\', '\\\\')
.replace(';', '\\;')
.replace(',', '\\,')
.replace('\n', '\\n')
.replace('\r', ''))
def _fold_line(line: str) -> str:
if len(line) <= 75:
return line
parts = [line[:73]]
rest = line[73:]
while rest:
parts.append(' ' + rest[:72])
rest = rest[72:]
return '\r\n'.join(parts)
def _dtstamp() -> str:
return datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
def _next_dow_date(days_of_week: List[int], today: date) -> date:
for offset in range(7):
candidate = today + timedelta(days=offset)
if (candidate.weekday() + 1) % 7 in days_of_week:
return candidate
return today
def _next_dom_date(days_of_month: List[int], today: date) -> date:
for offset in range(62):
candidate = today + timedelta(days=offset)
if candidate.day in days_of_month:
return candidate
return today
def _slugify(value: str) -> str:
return ''.join(c if c.isalnum() else '-' for c in (value or '').lower()).strip('-') or 'x'
def build_character_events(
character: Character,
site: SiteConfig,
*,
now: Optional[datetime] = None,
schedule_attr: str = 'articleSchedule',
kind: str = 'article',
summary_verb: str = 'Post',
) -> List[List[str]]:
sched = getattr(character, schedule_attr, None)
if sched is None:
return []
if not (sched.daysOfWeek or sched.daysOfMonth):
return []
today = (now or datetime.now()).date()
dtstamp = _dtstamp()
hours = sorted({h for h in (sched.preferredHours or []) if 0 <= h <= 23})
# Combine slug + short storageId suffix so distinct characters that happen
# to share a slug at the same site still get unique UIDs.
base = character.slug or character.name
sid = (character.storageId or '')[:8]
char_ident = f'{_slugify(base)}-{sid}' if sid else _slugify(base)
site_slug = _slugify(site.slug)
type_label = f'{kind.capitalize()} posting'
desc_parts = [
f'Type: {type_label}',
f'Character: {character.name}',
f'Site: {site.slug} ({site.wp_host})',
]
if hours:
window_end = hours[-1] + 1
desc_parts.append(
f'Posting window: {hours[0]:02d}:00\u2013{window_end:02d}:00 '
f'(one {kind} posted within this window)'
)
if character.description:
desc_parts.append(f'Description: {character.description.strip()[:200]}')
description = '\n'.join(desc_parts)
summary = f'{summary_verb}: {character.name} ({site.slug})'
blocks: List[List[str]] = []
def _emit(rrule: str, first_date: date, day_rule: str) -> None:
uid = f'tsai-schedule-{site_slug}-{char_ident}-{kind}-{day_rule}@tsai.local'
if not hours:
dtstart = f'DTSTART;VALUE=DATE:{first_date.strftime("%Y%m%d")}'
dtend = f'DTEND;VALUE=DATE:{(first_date + timedelta(days=1)).strftime("%Y%m%d")}'
else:
start = datetime.combine(first_date, datetime.min.time()).replace(hour=hours[0])
end = datetime.combine(first_date, datetime.min.time()) + timedelta(hours=hours[-1] + 1)
dtstart = f'DTSTART:{start.strftime("%Y%m%dT%H%M%S")}'
dtend = f'DTEND:{end.strftime("%Y%m%dT%H%M%S")}'
lines = [
'BEGIN:VEVENT',
f'UID:{uid}',
f'DTSTAMP:{dtstamp}',
dtstart,
dtend,
rrule,
f'SUMMARY:{_escape_text(summary)}',
f'DESCRIPTION:{_escape_text(description)}',
f'CATEGORIES:{kind.upper()}',
'END:VEVENT',
]
blocks.append(lines)
if sched.daysOfWeek:
valid_dow = sorted({d for d in sched.daysOfWeek if 0 <= d <= 6})
if valid_dow:
byday = ','.join(_BYDAY[d] for d in valid_dow)
_emit(f'RRULE:FREQ=WEEKLY;BYDAY={byday}', _next_dow_date(valid_dow, today), 'dow')
if sched.daysOfMonth:
valid_dom = sorted({d for d in sched.daysOfMonth if 1 <= d <= 31})
if valid_dom:
bymonthday = ','.join(str(d) for d in valid_dom)
_emit(
f'RRULE:FREQ=MONTHLY;BYMONTHDAY={bymonthday}',
_next_dom_date(valid_dom, today),
'dom',
)
return blocks
SCHEDULE_PASSES = (
('articleSchedule', 'article', 'Post'),
('commentSchedule', 'comment', 'Comment'),
)
# ~30-minute block for a concrete posted/failed event on the calendar.
_HISTORY_EVENT_MINUTES = 30
def build_history_events(
records: List[PostingRecord],
site: SiteConfig,
) -> List[List[str]]:
"""Concrete, dated VEVENTs for actual posting attempts (past days).
Unlike the forward-looking RRULE schedule events, each of these is a single
timed event carrying the real outcome: a clickable URL to the published
article/comment, the title, and success/failure status. Failures are marked
STATUS:CANCELLED so calendar apps grey/strike them out.
"""
site_slug = _slugify(site.slug)
dtstamp = _dtstamp()
blocks: List[List[str]] = []
for index, rec in enumerate(records):
try:
start = datetime.fromisoformat(rec.ts)
except (ValueError, TypeError):
continue
end = start + timedelta(minutes=_HISTORY_EVENT_MINUTES)
success = rec.status == 'success'
kind = rec.kind or 'post'
if success:
summary = (
f'✅ Posted: {rec.title}' if rec.title
else f'✅ {kind.capitalize()} posted: {rec.character}'
)
else:
summary = f'⚠️ Failed: {rec.character} ({kind})'
# Unique per attempt; post_id when present, else timestamp + index so
# multiple failures on one day don't collide.
ident = str(rec.post_id) if rec.post_id else f'{_slugify(rec.ts)}-{index}'
uid = f'tsai-post-{site_slug}-{kind}-{ident}@tsai.local'
desc_parts = [
f'Character: {rec.character}',
f'Site: {site.slug} ({site.wp_host})',
]
if rec.model:
desc_parts.append(f'Model: {rec.model}')
if rec.excerpt:
desc_parts.append(f'Excerpt: {rec.excerpt.strip()[:200]}')
if rec.url:
desc_parts.append(f'Link: {rec.url}')
if not success and rec.error:
desc_parts.append(f'Error: {rec.error}')
description = '\n'.join(desc_parts)
lines = [
'BEGIN:VEVENT',
f'UID:{uid}',
f'DTSTAMP:{dtstamp}',
f'DTSTART:{start.strftime("%Y%m%dT%H%M%S")}',
f'DTEND:{end.strftime("%Y%m%dT%H%M%S")}',
f'SUMMARY:{_escape_text(summary)}',
f'DESCRIPTION:{_escape_text(description)}',
f'STATUS:{"CONFIRMED" if success else "CANCELLED"}',
f'CATEGORIES:{kind.upper()},{"POSTED" if success else "FAILED"}',
]
# URL is a URI value — emitted unescaped (only folded).
if success and rec.url:
lines.append(f'URL:{rec.url}')
lines.append('END:VEVENT')
blocks.append(lines)
return blocks
def build_calendar(
pairs: List[Tuple[Character, SiteConfig]],
*,
calendar_name: str,
now: Optional[datetime] = None,
passes: Tuple[Tuple[str, str, str], ...] = SCHEDULE_PASSES,
history: Optional[List[Tuple[List[PostingRecord], SiteConfig]]] = None,
) -> str:
"""Assemble a VCALENDAR string from (character, site) pairs.
Deduplicates by (character identity, site.slug, kind) so a system-wide
character referenced under multiple sites still gets one set of events per
site, but article and comment events for the same character coexist.
`history` is an optional list of (posting records, site) pairs; each record
becomes a concrete dated event (past days) alongside the forward-looking
recurring schedule events.
"""
seen = set()
body: List[str] = []
for character, site in pairs:
char_key = character.storageId or character.slug or character.name
for schedule_attr, kind, summary_verb in passes:
key = (char_key, site.slug, kind)
if key in seen:
continue
seen.add(key)
blocks = build_character_events(
character, site,
now=now,
schedule_attr=schedule_attr,
kind=kind,
summary_verb=summary_verb,
)
for block in blocks:
body.extend(block)
for records, hist_site in (history or []):
for block in build_history_events(records, hist_site):
body.extend(block)
lines = [
'BEGIN:VCALENDAR',
'VERSION:2.0',
'PRODID:-//TSAI//Character Schedule Export//EN',
'CALSCALE:GREGORIAN',
'METHOD:PUBLISH',
f'X-WR-CALNAME:{_escape_text(calendar_name)}',
*body,
'END:VCALENDAR',
]
return '\r\n'.join(_fold_line(l) for l in lines) + '\r\n'