Spaces:
Paused
Paused
File size: 4,505 Bytes
ff36e71 | 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 | 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
@router.post("/fetch")
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()
|