1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import logging
import os
import sys
import time
from datetime import datetime, timezone

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# --- Logging Configuration ---
LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "attendance_api_final.log")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s.%(msecs)03d - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8", mode="a"),
        logging.StreamHandler(sys.stdout),
    ],
)

class SeiueAPIClient:
    """
    API client for Seiue to automate attendance via direct API calls.
    Uses a token-based auth flow.
    """
    def __init__(self, username: str, password: str):
        self.username = username
        self.password = password
        self.session = requests.Session()

        # Robust retries (includes 429 and common 5xx)
        retries = Retry(
            total=3,
            backoff_factor=1.5,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset({"HEAD", "GET", "POST", "PUT"}),
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retries))

        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
            "Accept": "application/json, text/plain, */*",
        })

        # --- API Endpoints ---
        self.login_url = "https://passport.seiue.com/login?school_id=3"
        self.authorize_url = "https://passport.seiue.com/authorize"
        self.events_url_template = "https://api.seiue.com/chalk/calendar/personals/{}/events"
        self.students_url_template = "https://api.seiue.com/scms/class/classes/{}/group-members?expand=reflection&member_type=student"
        self.attendance_submit_url_template = "https://api.seiue.com/sams/attendance/class/{}/records/sync"

        # --- Authentication Details ---
        self.bearer_token = None
        self.school_id = "3"
        self.role = "teacher"
        self.reflection_id = None

        # --- Safety & batching ---
        self.ALLOW_FUTURE_SUBMIT = False
        self.MAX_RECORDS_PER_REQUEST = 600

    # ---------- Helpers ----------
    def _re_auth(self) -> bool:
        logging.info("--- Re-authenticating (token refresh) ---")
        return self.login_and_get_token()

    def _with_refresh(self, request_fn):
        resp = request_fn()
        if getattr(resp, "status_code", None) == 401:
            logging.warning("401 received; attempting token refresh...")
            if self._re_auth():
                return request_fn()
        return resp

    @staticmethod
    def _parse_dt(s: str) -> datetime:
        # Convert API times to aware datetimes
        try:
            # ISO forms, convert 'Z' to +00:00 if present
            return datetime.fromisoformat(s.replace("Z", "+00:00"))
        except Exception:
            # Fallback: naive 'YYYY-MM-DD HH:MM:SS' -> assume UTC
            return datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)

    # ---------- Auth ----------
    def login_and_get_token(self) -> bool:
        logging.info("--- Starting Authentication Flow ---")
        try:
            logging.info("Step 1: Submitting login form...")
            login_form_data = {"email": self.username, "password": self.password}
            login_headers = {
                "Referer": self.login_url,
                "Origin": "https://passport.seiue.com",
                "Content-Type": "application/x-www-form-urlencoded",
            }
            login_resp = self.session.post(self.login_url, headers=login_headers, data=login_form_data, timeout=30)
            # Expect redirect into bindings or chalk
            if "bindings" not in login_resp.url and "chalk-c3" not in login_resp.url:
                logging.error(f"Unexpected post-login location: {login_resp.url}")
                return False
            logging.info(f"Login form submitted, redirected to: {login_resp.url}")

            logging.info("Step 2: Requesting access token...")
            auth_form_data = {"client_id": "GpxvnjhVKt56qTmnPWH1sA", "response_type": "token"}
            auth_headers = {
                "Referer": "https://chalk-c3.seiue.com/",
                "Origin": "https://chalk-c3.seiue.com",
                "Content-Type": "application/x-www-form-urlencoded",
            }
            auth_resp = self.session.post(self.authorize_url, headers=auth_headers, data=auth_form_data, timeout=30)
            auth_resp.raise_for_status()
            auth_data = auth_resp.json()

            self.bearer_token = auth_data.get("access_token")
            self.reflection_id = auth_data.get("active_reflection_id")
            if self.bearer_token and self.reflection_id:
                self.session.headers.update({
                    "Authorization": f"Bearer {self.bearer_token}",
                    "x-school-id": self.school_id,
                    "x-role": self.role,
                    "x-reflection-id": str(self.reflection_id),
                })
                logging.info("Token and reflection id acquired.")
                return True
            logging.error(f"Token/Reflection missing in response: {auth_data}")
            return False
        except Exception as e:
            logging.error(f"Authentication error: {e}", exc_info=True)
            return False

    # ---------- Data fetch ----------
    def get_lessons_to_attend(self, target_date: datetime):
        logging.info(f"Fetching lessons for date: {target_date.strftime('%Y-%m-%d')}")
        start_time_str = target_date.strftime("%Y-%m-%d 00:00:00")
        end_time_str = target_date.strftime("%Y-%m-%d 23:59:59")
        try:
            events_params = {"start_time": start_time_str, "end_time": end_time_str, "expand": "address,initiators"}
            events_url = self.events_url_template.format(self.reflection_id)
            events_resp = self._with_refresh(lambda: self.session.get(events_url, params=events_params, timeout=30))
            events_resp.raise_for_status()
            events = events_resp.json()
            logging.info(f"Found {len(events)} calendar events for this date.")

            if not events:
                logging.info(f"No scheduled calendar events found for {target_date.strftime('%Y-%m-%d')}.")
                return []

            now_utc = datetime.now(timezone.utc)
            lessons_to_attend = []
            lessons_already_attended_count = 0
            total_lesson_count = 0

            for lesson in events:
                if lesson.get("type") == "lesson":
                    total_lesson_count += 1
                    # Safety: avoid future classes unless explicitly enabled
                    try:
                        st = self._parse_dt(lesson.get("start_time"))
                    except Exception:
                        st = now_utc  # if parsing fails, allow processing to avoid silent skips
                    if (not self.ALLOW_FUTURE_SUBMIT) and (st > now_utc):
                        continue

                    if lesson.get("custom", {}).get("attendance_status") == 0:
                        logging.info(f"Pending attendance: {lesson.get('title')} @ {lesson.get('start_time')}")
                        lessons_to_attend.append(lesson)
                    else:
                        lessons_already_attended_count += 1

            if not lessons_to_attend and total_lesson_count > 0:
                logging.info(f"All {total_lesson_count} lessons for {target_date.strftime('%Y-%m-%d')} are already attended.")
            elif not lessons_to_attend:
                logging.info(f"No lessons requiring attendance on {target_date.strftime('%Y-%m-%d')}.")

            return lessons_to_attend
        except Exception as e:
            logging.error(f"Error fetching lessons: {e}", exc_info=True)
            return []

    def get_students_for_class(self, class_group_id: int):
        logging.info(f"Fetching student list for class group ID {class_group_id}...")
        url = self.students_url_template.format(class_group_id)
        try:
            resp = self._with_refresh(lambda: self.session.get(url, timeout=20))
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logging.error(f"Failed to fetch student list for class {class_group_id}: {e}")
            if hasattr(e, "response") and e.response is not None:
                logging.error(f"Status Code: {e.response.status_code}, Response: {e.response.text}")
            return None

    # ---------- Submit ----------
    def _submit_records_chunked(self, class_group_id: int, course_name: str, records: list) -> bool:
        submit_url = self.attendance_submit_url_template.format(class_group_id)
        for i in range(0, len(records), self.MAX_RECORDS_PER_REQUEST):
            payload = {
                "abnormal_notice_roles": [],  # make guardian/mentor an opt-in per-user policy
                "attendance_records": records[i : i + self.MAX_RECORDS_PER_REQUEST],
            }
            r = self._with_refresh(lambda: self.session.put(submit_url, json=payload, timeout=30))
            try:
                r.raise_for_status()
            except Exception:
                logging.error("Chunk %d failed (%s): %s", i // self.MAX_RECORDS_PER_REQUEST, r.status_code, r.text)
                return False
        logging.info("Submitted %d records for '%s' in %d chunk(s).",
                     len(records), course_name, (len(records) + self.MAX_RECORDS_PER_REQUEST - 1) // self.MAX_RECORDS_PER_REQUEST)
        return True

    def submit_attendance_for_lesson_group(self, lesson_group: list) -> bool:
        if not lesson_group:
            return False

        first_lesson = lesson_group[0]
        course_name = first_lesson.get("title", "Unknown Course")
        class_group_id = first_lesson.get("subject", {}).get("id")
        logging.info(f"--- Processing course group '{course_name}' ({len(lesson_group)} session(s)) ---")

        students = self.get_students_for_class(class_group_id)
        if students is None:
            return False
        if not students:
            logging.warning(f"Student list for course '{course_name}' is empty. Marking as complete.")
            return True

        all_attendance_records = []
        for lesson in lesson_group:
            attendance_time_id = lesson.get("custom", {}).get("id")
            if attendance_time_id is None:
                continue
            for student in students:
                ref = student.get("reflection", {}) or {}
                owner_id = ref.get("id")
                if owner_id:
                    all_attendance_records.append({
                        "tag": "正常",
                        "attendance_time_id": int(attendance_time_id),
                        "owner_id": owner_id,
                        "source": "web",
                    })

        if not all_attendance_records:
            logging.error("No valid attendance records were constructed from the student list.")
            return False

        if not self._submit_records_chunked(class_group_id, course_name, all_attendance_records):
            return False
        return True

# ---------- Orchestration ----------
def run_attendance_task(target_date_str: str) -> bool:
    username = os.getenv("SEIUE_USERNAME")
    password = os.getenv("SEIUE_PASSWORD")

    if not username or not password:
        logging.error("Username or password not set.")
        return False

    client = SeiueAPIClient(username, password)
    if not client.login_and_get_token():
        logging.error("Login and token acquisition failed.")
        logging.info(f"SUMMARY:{target_date_str}:status=failed_auth")
        return False

    target_date = datetime.strptime(target_date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    lessons = client.get_lessons_to_attend(target_date)

    if not lessons:
        logging.info(f"SUMMARY:{target_date_str}:status=no_events_or_no_pending")
        return True

    lessons_by_course = {}
    for lesson in lessons:
        course_id = lesson.get("subject", {}).get("id")
        if course_id is None:
            # Skip malformed lesson
            continue
        lessons_by_course.setdefault(course_id, []).append(lesson)

    all_groups_succeeded = True
    for _, lesson_group in lessons_by_course.items():
        ok = client.submit_attendance_for_lesson_group(lesson_group)
        if not ok:
            all_groups_succeeded = False
        time.sleep(2)

    # Post-submit verification
    verify = client.get_lessons_to_attend(target_date)
    pending_after = [l for l in verify if l.get("custom", {}).get("attendance_status") == 0]
    if pending_after:
        logging.warning("Post-submit verification found pending lessons still present.")
        all_groups_succeeded = False

    logging.info(f"SUMMARY:{target_date_str}:status={'ok' if all_groups_succeeded else 'failed'}")
    return all_groups_succeeded


if __name__ == "__main__":
    if len(sys.argv) < 2:
        logging.error("Execution Error: Please provide a date as a command-line argument (format: YYYY-MM-DD).")
        sys.exit(2)

    target_date_str = sys.argv[1]
    try:
        datetime.strptime(target_date_str, "%Y-%m-%d")
    except ValueError:
        logging.error(f"Invalid date format: '{target_date_str}'. Please use YYYY-MM-DD.")
        sys.exit(3)

    sys.exit(0 if run_attendance_task(target_date_str) else 1)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

#!/bin/bash
set -Eeuo pipefail

# --- Configuration ---
PROJECT_DIR="/Users/ylsuen/seiue" # Absolute path to the project directory
VENV_ACTIVATE="$PROJECT_DIR/venv/bin/activate"
PYTHON_EXECUTABLE="$PROJECT_DIR/venv/bin/python3"
PYTHON_SCRIPT="$PROJECT_DIR/APIFinal.py"
WRAPPER_LOG_FILE="$PROJECT_DIR/run_attendance_wrapper.log"
PING_TARGET="api.seiue.com"
LOCK_FILE="/tmp/seiue_attendance.lock"

# --- Telegram Configuration ---
TELEGRAM_BOT_TOKEN="${TELEGRAM_BOT_TOKEN:-}"
TELEGRAM_CHAT_ID="${TELEGRAM_CHAT_ID:-}"

# --- Utility Functions ---
log_message() {
  # BSD date lacks %N; prefer gdate when available, else use Python
  if command -v gdate >/dev/null 2>&1; then
    ts="$(gdate '+%Y-%m-%dT%H:%M:%S.%3N%z')"
  else
    ts="$(python3 - <<'PY'
from datetime import datetime, timezone
print(datetime.now(timezone.utc).astimezone().isoformat(timespec="milliseconds"))
PY
)"
  fi
  echo "$ts - $1" >> "$WRAPPER_LOG_FILE"
}

send_telegram_message() {
  local text="$1"
  if [[ -z "$TELEGRAM_BOT_TOKEN" || -z "$TELEGRAM_CHAT_ID" ]]; then
    log_message "WARNING: Telegram not configured; skipping."
    return 0
  fi
  # Plain-text JSON POST (avoid brittle URL query encoding)
  curl -sS -X POST "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/sendMessage" \
    -H 'Content-Type: application/json' \
    -d "{\"chat_id\":\"${TELEGRAM_CHAT_ID}\",\"text\":$(python3 -c 'import json,sys; print(json.dumps(sys.argv[1]))' "$text"),\"disable_web_page_preview\":true}" \
    >/dev/null || true
  log_message "Telegram notification sent."
}

check_network() {
  log_message "Checking network connectivity (Target: $PING_TARGET)..."
  if ping -c 1 -W 2000 "$PING_TARGET" >/dev/null 2>&1; then
    log_message "Network connection is OK."; return 0;
  else
    log_message "Network connection failed or target is unreachable."; return 1;
  fi
}

# --- Main Logic ---
# Simple lock to prevent overlapping runs
exec 9>"$LOCK_FILE" || true
if ! flock -n 9; then
  log_message "Another run is already in progress; exiting."
  exit 0
fi

log_message "--- Wrapper script started (API Mode | Daily Coverage Task) ---"
cd "$PROJECT_DIR" || { log_message "FATAL: Could not enter project directory."; exit 1; }
log_message "Current directory: $(pwd)"

# Activate virtual environment
if [ ! -f "$VENV_ACTIVATE" ]; then
  log_message "FATAL: Virtual environment not found at '$VENV_ACTIVATE'."; exit 1;
fi
source "$VENV_ACTIVATE"

# Check network
if ! check_network; then
  log_message "FATAL: Network check failed. Aborting."
  send_telegram_message "Seiue Attendance Bot: Network unreachable on $(hostname -s)."
  exit 1
fi

# --- Core Scheduling Logic ---
# Robust timezone handling (Beijing Time)
YESTERDAY=$(python3 -c "from datetime import datetime, timedelta; import pytz; tz=pytz.timezone('Asia/Shanghai'); print((datetime.now(tz) - timedelta(days=1)).strftime('%Y-%m-%d'))")
TODAY=$(python3 -c "from datetime import datetime; import pytz; tz=pytz.timezone('Asia/Shanghai'); print(datetime.now(tz).strftime('%Y-%m-%d'))")
TOMORROW=$(python3 -c "from datetime import datetime, timedelta; import pytz; tz=pytz.timezone('Asia/Shanghai'); print((datetime.now(tz) + timedelta(days=1)).strftime('%Y-%m-%d'))")

DATES_TO_PROCESS=("$YESTERDAY" "$TODAY" "$TOMORROW")
summary_lines=""
overall_success=true

send_telegram_message "[${HOSTNAME:-$(hostname -s)}] Starting Seiue Attendance Task..."

for date_str in "${DATES_TO_PROCESS[@]}"; do
  log_message "--- Processing date: $date_str ---"
  output="$("$PYTHON_EXECUTABLE" "$PYTHON_SCRIPT" "$date_str" 2>&1)"
  status=$?
  log_message "Output for $date_str:
$output"

  # Prefer stable SUMMARY tokens
  token="$(printf "%s\n" "$output" | grep -E '^SUMMARY:' | tail -n1 || true)"
  if [[ "$status" -eq 0 && -n "$token" ]]; then
    case "$token" in
      *status=ok*)               summary_lines+=$(printf "✅ %s: OK\n" "$date_str");;
      *status=failed_auth*)      summary_lines+=$(printf "❌ %s: Auth failed\n" "$date_str"); overall_success=false;;
      *status=no_events_or_no_pending*) summary_lines+=$(printf "✅ %s: No events / no pending\n" "$date_str");;
      *status=failed*)           summary_lines+=$(printf "❌ %s: Failed\n" "$date_str"); overall_success=false;;
      *)                         summary_lines+=$(printf "✅ %s: Completed\n" "$date_str");;
    esac
  else
    summary_lines+=$(printf "❌ %s: Exit %s\n" "$date_str" "$status")
    overall_success=false
  fi
done

# --- Final Reporting ---
CURRENT_TIME_CST=$(python3 -c "from datetime import datetime; import pytz; print(datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S'))")
final_message=$(printf "[Seiue Attendance Summary]\n\n%s\nTimestamp (CST): %s" "$summary_lines" "$CURRENT_TIME_CST")

log_message "Final summary prepared. Overall success: $overall_success"
send_telegram_message "$final_message"

log_message "--- Wrapper script finished ---"

if $overall_success; then
  exit 0
else
  exit 1
fi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.ylsuen.seiue.attendance.api</string>

    <key>ProgramArguments</key>
    <array>
        <string>/bin/bash</string>
        <string>/Users/ylsuen/seiue/run_attendance.sh</string>
    </array>

    <key>StartCalendarInterval</key>
    <dict>
        <!-- This is your Mac's local time. -->
        <!-- Set to 5 AM to correspond to 8 PM in Beijing (PDT). -->
        <key>Hour</key>
        <integer>5</integer>
        <key>Minute</key>
        <integer>0</integer>
    </dict>

    <key>WorkingDirectory</key>
    <string>/Users/ylsuen/seiue</string>

    <key>StandardOutPath</key>
    <string>/Users/ylsuen/seiue/launchd_api.log</string>

    <key>StandardErrorPath</key>
    <string>/Users/ylsuen/seiue/launchd_api.error.log</string>
    
    <key>EnvironmentVariables</key>
    <dict>
        <key>PATH</key>
        <string>/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/opt/homebrew/bin:/Users/ylsuen/seiue/venv/bin</string>
    </dict>

    <key>RunAtLoad</key>
    <false/>

    <key>Disabled</key>
    <false/>

</dict>
</plist>

1
2

/opt/homebrew/bin/python3 -m venv /Users/ylsuen/seiue/venv
/Users/ylsuen/seiue/venv/bin/pip install -r /Users/ylsuen/seiue/requirements.txt