|
| 1 | +import urllib.parse |
| 2 | +from user_scanner.core.helpers import get_random_user_agent |
| 3 | +from user_scanner.core.orchestrator import Result, generic_validate |
| 4 | + |
| 5 | +def validate_archesproject(user: str) -> Result: |
| 6 | + encoded_user = urllib.parse.quote(user) |
| 7 | + url = f"https://community.archesproject.org/u/{encoded_user}.json" |
| 8 | + show_url = f"https://community.archesproject.org/u/{encoded_user}" |
| 9 | + |
| 10 | + headers = { |
| 11 | + "User-Agent": get_random_user_agent(), |
| 12 | + "Accept": "application/json", |
| 13 | + } |
| 14 | + |
| 15 | + def process(response) -> Result: |
| 16 | + if response.status_code == 404: |
| 17 | + try: |
| 18 | + data = response.json() |
| 19 | + if ( |
| 20 | + data.get("error_type") == "not_found" |
| 21 | + or any("not be found" in str(err).lower() for err in data.get("errors", [])) |
| 22 | + ): |
| 23 | + return Result.available(url=show_url) |
| 24 | + except Exception: |
| 25 | + pass |
| 26 | + return Result.error("Discourse 404 response missing expected error payload") |
| 27 | + |
| 28 | + if response.status_code == 200: |
| 29 | + try: |
| 30 | + data = response.json() |
| 31 | + user_obj = data.get("user") |
| 32 | + if user_obj and isinstance(user_obj, dict) and user_obj.get("username"): |
| 33 | + extra: dict[str, str] = {} |
| 34 | + media: dict[str, str] = {} |
| 35 | + |
| 36 | + if user_obj.get("name"): |
| 37 | + extra["name"] = str(user_obj["name"]) |
| 38 | + if user_obj.get("title"): |
| 39 | + extra["title"] = str(user_obj["title"]) |
| 40 | + if user_obj.get("location"): |
| 41 | + extra["location"] = str(user_obj["location"]) |
| 42 | + if user_obj.get("website_name") or user_obj.get("website"): |
| 43 | + extra["website"] = str(user_obj.get("website") or user_obj.get("website_name")) |
| 44 | + if user_obj.get("trust_level") is not None: |
| 45 | + extra["trust_level"] = str(user_obj["trust_level"]) |
| 46 | + if user_obj.get("badge_count") is not None: |
| 47 | + extra["badges"] = str(user_obj["badge_count"]) |
| 48 | + if user_obj.get("created_at"): |
| 49 | + extra["joined"] = str(user_obj["created_at"]) |
| 50 | + if user_obj.get("bio_raw") or user_obj.get("bio_cooked"): |
| 51 | + extra["bio"] = str(user_obj.get("bio_raw") or user_obj.get("bio_cooked")) |
| 52 | + |
| 53 | + avatar_template = user_obj.get("avatar_template") |
| 54 | + if avatar_template: |
| 55 | + avatar_url = avatar_template.replace("{size}", "240") |
| 56 | + if avatar_url.startswith("/"): |
| 57 | + avatar_url = f"https://community.archesproject.org{avatar_url}" |
| 58 | + media["avatar"] = avatar_url |
| 59 | + |
| 60 | + return Result.taken(url=show_url, extra=extra, media=media) |
| 61 | + except Exception: |
| 62 | + pass |
| 63 | + return Result.error("Discourse 200 response missing valid user payload") |
| 64 | + |
| 65 | + return Result.error(f"Unexpected status code: {response.status_code}") |
| 66 | + |
| 67 | + return generic_validate(url, process, headers=headers, show_url=show_url, follow_redirects=True) |
0 commit comments