403Webshell
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 :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /tsai/repo/api/app/services/schedule_calendar.py
"""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'

Youez - 2016 - github.com/yon3zu
LinuXploit