Spaces:
Paused
Paused
| from fastapi import APIRouter, HTTPException | |
| from pydantic import BaseModel | |
| import re | |
| from typing import List, Optional | |
| import base64 | |
| router = APIRouter() | |
| class IServCredentials(BaseModel): | |
| url: str | |
| username: str | |
| password: str | |
| def censor(text: str, username: str) -> str: | |
| if not text: | |
| return "" | |
| # Simple regex for emails | |
| email_regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' | |
| phone_regex = r'\+?49[\s\-]?\(?\d{2,4}\)?[\s\-]?\d{3,8}' | |
| clean_text = re.sub(email_regex, '<EMAIL>', text) | |
| clean_text = re.sub(phone_regex, '<TELEFON>', clean_text) | |
| if username and '.' in username: | |
| parts = username.split('@')[0].split('.') | |
| for part in parts: | |
| if len(part) > 2: | |
| clean_text = re.sub(re.escape(part), '<NAME>', clean_text, flags=re.IGNORECASE) | |
| return clean_text | |
| async def fetch_iserv_tasks(creds: IServCredentials): | |
| """ | |
| Fetches tasks from IServ using Playwright. | |
| Note: Requires 'playwright install chromium' to be run on the server. | |
| """ | |
| try: | |
| from playwright.async_api import async_playwright | |
| except ImportError: | |
| raise HTTPException(status_code=500, detail="Playwright is not installed on the server.") | |
| url = creds.url | |
| if not url.startswith('http'): | |
| url = 'https://' + url | |
| exercise_url = f"{url}/iserv/exercise" | |
| async with async_playwright() as p: | |
| browser = await p.chromium.launch(headless=True) | |
| try: | |
| context = await browser.new_context() | |
| page = await context.new_page() | |
| await page.goto(exercise_url) | |
| # Login | |
| if '/login' in page.url: | |
| await page.fill("input[name='_username']", creds.username) | |
| await page.fill("input[name='_password']", creds.password) | |
| submit_btn = await page.query_selector("button[type='submit'], input[type='submit'], button.btn-primary") | |
| if submit_btn: | |
| await submit_btn.click() | |
| else: | |
| await page.evaluate("document.querySelector('form').submit()") | |
| try: | |
| await page.wait_for_load_state("networkidle", timeout=10000) | |
| except: | |
| pass | |
| # Check for error | |
| if 'authentication/error' in page.url: | |
| raise Exception("Login failed. Check credentials.") | |
| # Ensure we are on exercise page | |
| if '/iserv/exercise' not in page.url: | |
| await page.goto(exercise_url) | |
| try: | |
| await page.wait_for_selector("table.table tbody tr", timeout=5000) | |
| rows = await page.query_selector_all("table.table tbody tr") | |
| except: | |
| try: | |
| await page.wait_for_selector("article, .card", timeout=5000) | |
| rows = await page.query_selector_all("article, .card") | |
| except: | |
| rows = [] | |
| tasks = [] | |
| for row in rows: | |
| link_el = await row.query_selector("td a") | |
| if not link_el: | |
| continue | |
| title = await link_el.inner_text() | |
| href = await link_el.get_attribute("href") | |
| full_url = f"{url}{href}" if href.startswith("/") else href | |
| cells = await row.query_selector_all("td") | |
| deadline = "" | |
| if len(cells) > 2: | |
| deadline = await cells[2].inner_text() | |
| tasks.append({ | |
| "id": full_url or title.strip(), | |
| "title": censor(title.strip(), creds.username), | |
| "deadline": deadline.strip(), | |
| "url": full_url, | |
| "description": "", | |
| "attachments": [] | |
| }) | |
| # Details fetching omitted for brevity/speed on HF spaces, | |
| # but can be added back by looping through tasks and visiting full_url | |
| return tasks | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| finally: | |
| await browser.close() | |