diff --git a/.env b/.env new file mode 100644 index 0000000..82f4846 --- /dev/null +++ b/.env @@ -0,0 +1,4 @@ +# Environment variables for profile-checker +# Available levels: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET +LOG_LEVEL=DEBUG +LOG_FILE=finder.log diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml index 9c8a85d..1f8ede8 100644 --- a/.github/workflows/pylint.yml +++ b/.github/workflows/pylint.yml @@ -17,6 +17,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip + pip install -r requirements.txt pip install pylint - name: Analysing the code with pylint run: | diff --git a/README.md b/README.md index e0d74e7..c19333e 100644 --- a/README.md +++ b/README.md @@ -1,78 +1,52 @@ -# scan -scan is an OSINT microservice that searches the [WhatsMyName](https://github.com/WebBreacher/WhatsMyName) dataset for username hits from various websites. -The service uses a database to cache search results because validating usernames is costly and time-consuming. Cached results are reused on repeat lookups, and a refresh option exists to rebuild the results. -A status endpoint is exposed so clients can poll long-running searches instead of relying on a single request timeout. +# profile-checker + +Profile validation microservice called `profile-checker` that checks whether usernames exist on social media platforms. The service validates profiles by performing HTTP requests to platform-specific endpoints and confirms existence through response validation. ## Requirements - Python 3.9 or higher ### Tech Stack - The REST API framework being used is [FastAPI](https://fastapi.tiangolo.com/) -- The database engine is [SQLite](https://www.sqlite.org/index.html) - The REST client being used is [httpx](https://www.python-httpx.org/) ### Setup -1. Create Python virtual environment. `python -m venv venv` -2. Activate virtual environment. `source venv/bin/activate` +1. Create Python virtual environment. `python -m venv .venv` +2. Activate virtual environment. `source .venv/bin/activate` 3. Install dependencies. `pip install -r requirements.txt` -4. Start server. `fastapi dev main.py` +4. Create a `.env` file with `LOG_LEVEL` and `LOG_FILE` as needed. +5. Start server. `fastapi dev main.py` + +### Configuration +- `config.py` initializes logging and reads environment variables. +- `.env` supports: + - `LOG_LEVEL` — log verbosity (`DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`) + - `LOG_FILE` — path to the log output file +- Site verification templates are stored in `sites.yaml`. ### Endpoints - GET `/scan/{username}` - * Starts a new search if the username has not been cached yet. - * Returns `202 Accepted` when a new background search is started. - * Returns `102 Processing` when a search for the username is already pending or in progress. - * Returns cached results with `200 OK` if the username has already been searched and no refresh is requested. - * Query Parameters: - * `refresh` (`true` or `false`, default `false`) — when set to `true`, the cache is invalidated and the username search is rebuilt. + * Validates whether the given username exists on supported platforms. + * Returns `200 OK` with results containing found profiles. + * Performs HEAD requests to platform validation endpoints and follows up with GET requests to confirm profile existence. -- GET `/scan/status/{username}` - * Returns the current status of a background search. - * Use this endpoint to poll progress until the search reaches `completed`. - -### Example response for `/scan/{username}` when cached +### Example response for `/scan/{username}` ```json [ { - "title": "GitHub", - "uri": "https://github.com/osint-services", - "profile_url": "https://github.com/osint-services", - "category": "coding", - "search_timestamp": "2024-11-02 17:01:27", - "found_timestamp": "2024-11-02 17:01:28" - }, - { - "title": "Facebook", - "uri": "https://facebook.com/osint-services", - "profile_url": "https://facebook.com/osint-services", - "category": "social", - "search_timestamp": "2024-11-02 17:01:27", - "found_timestamp": "2024-11-02 17:01:28" + "title": "X", + "profile_uri": "https://x.com/osint-services", + "validation_uri": "https://api.x.com/i/users/username_available.json?username=osint-services", + "is_valid_profile": true } ] ``` -### Example response for `/scan/status/{username}` -```json -{ - "status": "in_progress", - "found_sites": [ - { - "title": "GitHub", - "uri": "https://github.com/osint-services", - "profile_url": "https://github.com/osint-services", - "category": "coding", - "search_timestamp": "2024-11-02 17:01:27", - "found_timestamp": "2024-11-02 17:01:28" - } - ], - "total_sites": 200, - "checked_sites": 50 -} -``` - ### Notes -- This service uses background processing and polling rather than long-lived request timeouts. -- The result validation step checks page content and common profile signals to reduce false positives from login, not-found, or generic landing pages. -- If a search fails, `GET /scan/status/{username}` will return a `failed` status and an `error` field may be included in the response. +- Profile validation checks page responses and validation endpoints to confirm actual profile existence and reduce false positives. +- The service currently supports X (Twitter) profile validation. +- Logging is configured in `config.py` and writes to both console and the file specified by `LOG_FILE`. +- Supported site templates are defined in `sites.yaml`, making it easy to add or update platform endpoints without changing application logic. +- `LOG_LEVEL` and `LOG_FILE` can be configured via `.env`. +- The project root already includes `__init__.py`, so the package can be imported if needed, but script execution does not require additional `__init__.py` files. +- CORS is enabled for localhost:3000 frontend connections. diff --git a/config.py b/config.py new file mode 100644 index 0000000..dee3bdc --- /dev/null +++ b/config.py @@ -0,0 +1,30 @@ +"""Application configuration and logging setup.""" + +import logging +import os + +from dotenv import load_dotenv + +load_dotenv() + +LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG").upper() +if LOG_LEVEL not in {"CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"}: + LOG_LEVEL = "DEBUG" + +LOG_FILE = os.getenv("LOG_FILE", "finder.log") + +logger = logging.getLogger("profile-checker") +logger.setLevel(LOG_LEVEL) + +ch = logging.StreamHandler() +ch.setLevel(LOG_LEVEL) + +fh = logging.FileHandler(LOG_FILE) +fh.setLevel(LOG_LEVEL) + +formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") +ch.setFormatter(formatter) +fh.setFormatter(formatter) + +logger.addHandler(ch) +logger.addHandler(fh) diff --git a/database.py b/database.py deleted file mode 100644 index f8ca0e2..0000000 --- a/database.py +++ /dev/null @@ -1,155 +0,0 @@ -import sqlite3 -import logging - -# Connect to the SQLite database (or create it if it doesn't exist) -conn = sqlite3.connect('whatsmyname.db', check_same_thread=False) - -# Set up logger -logger = logging.getLogger(__name__) -logger.setLevel(logging.INFO) # You can change to DEBUG or ERROR based on your needs - -def has_username_been_searched(username): - try: - logger.debug(f"Checking cache for username '{username}'") - cursor = conn.cursor() - query = """ - SELECT 1 - FROM usernames_searched - WHERE username = ?; - """ - cursor.execute(query, (username,)) - result = cursor.fetchone() - if result: - logger.info(f"Username '{username}' cache hit.") - return True - logger.info(f"Username '{username}' not found in cache.") - return False - except sqlite3.DatabaseError as e: - logger.exception("Database error in has_username_been_searched") - raise - - -def insert_username(username): - try: - cursor = conn.cursor() - query = """ - INSERT INTO usernames_searched (username) - VALUES (?); - """ - cursor.execute(query, (username,)) - conn.commit() - logger.info(f"Username '{username}' has been inserted.") - except sqlite3.IntegrityError: - logger.warning(f"Username '{username}' already exists.") - except sqlite3.DatabaseError as e: - logger.error(f"Database error in insert_username: {e}") - raise e - - -def insert_username_correlation(username, site): - try: - cursor = conn.cursor() - cursor.execute('SELECT id FROM usernames_searched WHERE username = ?', (username,)) - username_id = cursor.fetchone() - - cursor.execute('SELECT id FROM sites WHERE name = ? AND uri_check = ?', (site['title'], site['uri'])) - wm_data_id = cursor.fetchone() - - if username_id and wm_data_id: - cursor.execute(''' - INSERT INTO username_correlations (username_id, site_id) - VALUES (?, ?) - ''', (username_id[0], wm_data_id[0])) - conn.commit() - logger.info(f"Correlation for username '{username}' and site '{site['uri']}' inserted.") - else: - logger.warning(f"No correlation found for username '{username}' and site '{site['uri']}'.") - except sqlite3.DatabaseError as e: - logger.error(f"Database error in insert_username_correlation: {e}") - raise e - - -def insert_sites(sites: list[dict]): - try: - cursor = conn.cursor() - cursor.executemany(''' - INSERT INTO sites (name, uri_check, cat) - VALUES (:name, :uri_check, :cat); - ''', sites) - conn.commit() - logger.info(f"Inserted {len(sites)} sites into the database.") - except sqlite3.DatabaseError as e: - logger.error(f"Database error in insert_sites: {e}") - raise e - - -def get_all_sites(): - try: - cursor = conn.cursor() - cursor.execute('SELECT name, uri_check, cat FROM sites') - def serialize_site(site: list) -> dict: - return { - "title": site[0], - "uri": site[1], - "category": site[2], - } - sites = cursor.fetchall() - logger.debug(f"Loaded {len(sites)} site definitions from the database") - return [serialize_site(site) for site in sites] - except sqlite3.DatabaseError as e: - logger.exception("Database error in get_all_sites") - raise - - -def get_sites_by_username(username): - try: - query = """ - SELECT s.name, s.uri_check, s.cat, u.search_timestamp, uc.found_timestamp - FROM sites s - JOIN username_correlations uc ON s.id = uc.site_id - JOIN usernames_searched u ON u.id = uc.username_id - WHERE u.username = ?; - """ - cursor = conn.cursor() - cursor.execute(query, (username,)) - sites = cursor.fetchall() - logger.info(f"Returning {len(sites)} cached matches for username '{username}'") - - def serialize_site(site: list) -> dict: - profile_url = site[1].format(account=username) - return { - "title": site[0], - "uri": profile_url, - "profile_url": profile_url, - "category": site[2], - "search_timestamp": site[3], - "found_timestamp": site[4] - } - - return [serialize_site(site) for site in sites] - except sqlite3.DatabaseError as e: - logger.exception("Database error in get_sites_by_username") - raise - -def delete_search_history(username: str): - try: - cursor = conn.cursor() - cursor.execute(''' - SELECT id FROM usernames_searched WHERE username = ?; - ''', (username,)) - user_id = cursor.fetchone() - if not user_id: - logger.info(f"No cache entry found for username '{username}' to delete") - return - - user_id = user_id[0] - cursor.execute(''' - DELETE FROM username_correlations WHERE username_id = ?; - ''', (user_id,)) - cursor.execute(''' - DELETE FROM usernames_searched WHERE id = ?; - ''', (user_id,)) - conn.commit() - except sqlite3.DatabaseError as e: - logger.exception("Database error in delete_search_history") - raise \ No newline at end of file diff --git a/main.py b/main.py index c9f06b5..82e6f64 100644 --- a/main.py +++ b/main.py @@ -1,350 +1,186 @@ -import httpx -import logging -import re -import os +""" +Profile Validation Service +""" -from fastapi import FastAPI, BackgroundTasks, status, HTTPException -from fastapi.responses import JSONResponse -from fastapi.middleware.cors import CORSMiddleware +from contextlib import asynccontextmanager from http import HTTPStatus +from pathlib import Path from ssl import SSLError -from anthropic import Anthropic -from dotenv import load_dotenv - -from .database import * - -load_dotenv() - -origins = [ - "http://localhost:3000" -] - -# Set up logger -logger = logging.getLogger(__name__) -logger.setLevel(logging.DEBUG) # You can change to DEBUG or ERROR based on your needs +import httpx +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from pydantic import ValidationError +import yaml -# Create console handler and set level to INFO -ch = logging.StreamHandler() -ch.setLevel(logging.DEBUG) +from .config import logger +from .models import SiteData, SiteResult, XValidationResponse, XUsernameAvailabilityReason -# Create file handler to store logs -fh = logging.FileHandler('finder.log') -fh.setLevel(logging.DEBUG) +origins = ["http://localhost:3000"] -# Create formatter and add it to the handlers -formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') -ch.setFormatter(formatter) -fh.setFormatter(formatter) +client = httpx.AsyncClient() -# Add the handlers to the logger -logger.addHandler(ch) -logger.addHandler(fh) +@asynccontextmanager +async def lifespan(app: FastAPI): + logger.info("Starting application lifecycle") + yield + logger.info("Shutting down HTTPX client") + await client.aclose() -app = FastAPI() +app = FastAPI(lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, - allow_methods=['*'], - allow_headers=['*'] + allow_methods=["*"], + allow_headers=["*"], ) -client = httpx.AsyncClient() -api_key = os.getenv("ANTHROPIC_API_KEY") -logger.info(f"Anthropic API Key: {'set' if api_key else 'not set'}") -ai_client = Anthropic(api_key=api_key) +def load_site_config(config_path: Path) -> list[dict]: + if not config_path.exists(): + raise FileNotFoundError(f"Site config not found: {config_path}") -@app.on_event("shutdown") -async def shutdown() -> None: - logger.info("Shutting down HTTPX client") - await client.aclose() + with config_path.open("r", encoding="utf-8") as fh: + config = yaml.safe_load(fh) -async def validate_profile_with_ai(url: str, username: str, html_snippet: str) -> bool: - try: - message = ai_client.messages.create( - model="claude-sonnet-4-6", - max_tokens=100, - messages=[ - { - "role": "user", - "content": f"""Analyze this web page HTML and determine if it is a real, valid profile page for the username '{username}'. + if not config or "sites" not in config: + raise ValueError("Site config must contain a top-level 'sites' key") -URL: {url} -Username: {username} + return config["sites"] -HTML snippet (first 2000 chars): -{html_snippet[:2000]} +SITE_CONFIG = load_site_config(Path(__file__).resolve().parent / "sites.yaml") -Respond with ONLY 'YES' if this appears to be a genuine profile page for the given username, or 'NO' if it does not. -Consider: -- Does the page seem to be a login, error, or generic landing page? -- Does the page contain content associated with a user profile (posts, followers, bio, etc)? -- Is the username visible or referenced on the page? -Respond with exactly 'YES' or 'NO'.""" - } - ] +def get_site_list(username: str) -> list[SiteData]: + """ + Returns a list of site data objects for the given username by applying the configured templates. + """ + return [ + SiteData( + title=site["title"], + profile_uri=site["profile_uri"].format(username=username), + validation_uri=site["validation_uri"].format(username=username), ) - - # Extract text from the first TextBlock in the response - response_text = "" - for block in message.content: - if hasattr(block, "text"): - response_text = block.text.strip().upper() # type: ignore - break - - is_valid = response_text == "YES" - logger.debug(f"AI validation for {url}: {response_text}") - return is_valid - except Exception as e: - logger.warning(f"AI validation failed for {url}: {e}") - return False + for site in SITE_CONFIG + ] + async def confirm_profile_exists(url: str, username: str, title: str) -> bool: + """ + For some sites, a HEAD request may return 200 OK for both existing and non-existing profiles. + In such cases, we need to perform a GET request to confirm the existence of the profile. + This function handles that logic, including error handling for various exceptions that may + occur during the HTTP request. + """ try: response = await client.get(url, follow_redirects=True, timeout=15.0) - except (httpx.ReadTimeout, httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadError, httpx.TooManyRedirects, ValueError, SSLError) as e: + except ( + httpx.ReadTimeout, + httpx.ConnectError, + httpx.ConnectTimeout, + httpx.ReadError, + httpx.TooManyRedirects, + ValueError, + SSLError, + ) as e: logger.warning(f"Profile validation failed for '{username}' at '{url}': {e}") return False except Exception as e: - logger.exception(f"Unexpected error during profile validation for '{username}' at '{url}'") + logger.exception( + f"Unexpected error during profile validation for '{username}' at '{url}': {e}" + ) return False if response.status_code != HTTPStatus.OK: - logger.debug(f"Profile validation GET returned {response.status_code} for {url}") - return False - - - # X returns an object that has a valid marker on it, if it's not valid - logger.info(f"Checking profile existence for '{username}' on {url} with title '{title}'") - if title.lower() == 'x': - username_available = response.json().get('valid') - reason = response.json().get('reason', 'No reason provided').lower() - return not username_available and reason == 'taken' # if username not available because it's taken then profile exists, if it's not available because it's invalid then profile doesn't exist - - final_url = str(response.url).lower() - deny_url_tokens = ["login", "signin", "sign-in", "sign_up", "register", "auth", "account", "signup"] - if any(token in final_url for token in deny_url_tokens): - logger.debug(f"Profile validation ignored redirect/login URL for {url}: {final_url}") - return False - - content_type = response.headers.get("content-type", "").lower() - if "html" not in content_type and "text" not in content_type: - logger.debug(f"Profile validation skipped non-HTML content for {url}: {content_type}") - return False - - body = response.text.lower() - deny_body_patterns = [ - "log in", "login", "sign in", "sign up", "register", "create account", - "not found", "page not found", "404", "410", "private", "forbidden", "access denied", - "account does not exist", "user not found", "profile does not exist", "this account is not available", - "there isn\'t anything here", "page isn\'t available", "this page isn\'t available", "this blog doesn\'t exist.", - "member not found", "couldn\'t find", "profile unavailable", "does not exist", - ] - - # Garmin has a specific blank page for non-existent profiles that doesn't include typical error phrases, so we can add a specific check for that - if '' in body and "garmin" in final_url: - logger.debug(f"Profile validation denied Garmin profile {url} due to signed-out blank page") - return False - - if any(pattern in body for pattern in deny_body_patterns): - logger.debug(f"Profile validation denied {url} due to body content") + logger.debug( + f"Profile validation GET returned {response.status_code} for {url}" + ) return False - title_match = re.search(r"]*>(.*?)<\/title>", response.text, re.IGNORECASE | re.DOTALL) - page_title = title_match.group(1).strip().lower() if title_match else "" - - og_type_match = re.search(r"]+property=[\"']og:type[\"'][^>]+content=[\"']([^\"']+)[\"']", response.text, re.IGNORECASE) - if og_type_match and "profile" in og_type_match.group(1).lower(): - logger.debug(f"Profile validation accepted {url} due to og:type profile") - return True - - username_lower = username.lower() - username_in_title = username_lower in page_title - username_in_url = username_lower in final_url - username_in_body = username_lower in body - - profile_markers = [ - "followers", "following", "posts", "timeline", "about", "bio", "overview", "activity", - "repositories", "repos", "projects", "connections", "friends", "photos", "videos", - "likes", "stories", "publications", "recent posts", "reviews", "blogs", "posts", "tweets" - ] - marker_count = sum(1 for marker in profile_markers if marker in body or marker in page_title) - - if username_in_title: - logger.debug(f"Profile validation accepted {url} because username appears in title") - return True - - if username_in_body and marker_count >= 1: - logger.debug(f"Profile validation accepted {url} because username and profile markers were found") - return True - - if username_in_url and marker_count >= 1: - logger.debug(f"Profile validation accepted {url} because URL and profile markers were found") - return True - - logger.debug(f"Profile heuristics inconclusive for {url}, escalating to AI validation") - is_profile = await validate_profile_with_ai(url, username, response.text) - return is_profile - -""" -Gets list of websites from WhatsMyName, this is used to search usernames -""" -def get_site_list() -> list[dict]: - url = "https://raw.githubusercontent.com/WebBreacher/WhatsMyName/main/wmn-data.json" - response = httpx.get(url, timeout=20.0) - - try: - if response.status_code == HTTPStatus.OK: - data = response.json() - logger.debug("Successfully retrieved site list from WhatsMyName") - return data['sites'] - logger.error(f"Failed to retrieve JSON. Status code: {response.status_code}") - raise Exception(f"Failed to retrieve JSON. Status code: {response.status_code}") - except Exception as e: - logger.exception("Failed to retrieve WhatsMyName dataset") - raise - -@app.on_event("startup") -async def boot(): - cursor = conn.cursor() - logger.info('Building database tables...') - # Create the tables - cursor.execute(''' - CREATE TABLE IF NOT EXISTS usernames_searched ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT UNIQUE, - search_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP - ); - ''') - - cursor.execute(''' - CREATE TABLE IF NOT EXISTS sites ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - uri_check TEXT NOT NULL, - cat TEXT NOT NULL, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP - ); - ''') - - cursor.execute(''' - CREATE TABLE IF NOT EXISTS username_correlations ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username_id INTEGER, - site_id INTEGER, - found_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (username_id) REFERENCES usernames_searched(id), - FOREIGN KEY (site_id) REFERENCES sites(id) - ); - ''') - conn.commit() - - logger.info('Ingesting website data...') - - try: - sites = get_site_list() - insert_sites(sites) - logger.info(f"Successfully inserted {len(sites)} sites into the database.") - except Exception as e: - logger.error("Failed to ingest website data") - logger.exception(e) - -task_status: dict[str, dict] = {} - -async def search_for_username(username: str) -> list: - sites_found = [] - insert_username(username) - logger.info(f"Starting background search for username '{username}'") - - sites = get_all_sites() - task_status[username] = { - "status": "in_progress", - "found_sites": [], - "total_sites": len(sites), - "checked_sites": 0, - } + logger.info( + f"Checking profile existence for '{username}' on {url} with title '{title}'" + ) + # map out responses using pydantic and validate the response structure, then determine if profile exists based on the response data + if title.lower() == "x": + try: + validation = XValidationResponse.parse_obj(response.json()) + except (ValidationError, ValueError) as e: + logger.warning( + f"Invalid validation response for '{username}' on '{url}': {e}" + ) + return False + + # if username is not available because it's taken then profile exists + return not validation.valid and validation.reason == XUsernameAvailabilityReason.taken + + return False + +async def search_for_username(username: str) -> list[SiteResult]: + """ + Searches for the given username across multiple sites by performing HEAD requests to the + validation URIs. If a HEAD request returns 200 OK, it performs a GET request to confirm + the existence of the profile. The function handles various exceptions that may occur during + HTTP requests and logs relevant information throughout the process. + """ + sites_found: list[SiteResult] = [] + sites = get_site_list(username) for site_data in sites: - uri = site_data['uri'] + validation_uri = site_data.validation_uri try: - response = await client.head(uri.format(account=username), follow_redirects=True) - task_status[username]["checked_sites"] += 1 - - if 'valid' in site_data and not site_data['valid']: - logger.debug(f"Skipping invalid site entry for '{username}': {uri}") - continue + response = await client.head(validation_uri, follow_redirects=True) if response.status_code == HTTPStatus.OK: - profile_url = uri.format(account=username) - is_profile = await confirm_profile_exists(profile_url, username, site_data.get('title')) + is_profile = await confirm_profile_exists( + validation_uri, username, site_data.title + ) if not is_profile: - logger.debug(f"Head-only match rejected for '{username}' on {profile_url}") + logger.debug( + f"Head-only match rejected for '{username}' on {validation_uri}" + ) continue - insert_username_correlation(username, site_data) - site_result = site_data.copy() - site_result["profile_url"] = profile_url - site_result["uri"] = profile_url + site_result = SiteResult( + title=site_data.title, + profile_uri=site_data.profile_uri, + validation_uri=site_data.validation_uri, + is_valid_profile=is_profile, + ) sites_found.append(site_result) - task_status[username]["found_sites"].append(site_result) - logger.debug(f"Username '{username}' found on site: {uri}") + logger.debug(f"Username '{username}' found on site: {validation_uri}") else: - logger.debug(f"No match for '{username}' on site: {uri} (status={response.status_code})") - except (httpx.ReadTimeout, httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadError, httpx.TooManyRedirects, ValueError, SSLError) as e: - logger.warning(f"Request failed for '{username}' on site '{uri}': {e}") + logger.debug(f"No match for '{username}' on site: {validation_uri} \ + (status={response.status_code})") + except ( + httpx.ReadTimeout, + httpx.ConnectError, + httpx.ConnectTimeout, + httpx.ReadError, + httpx.TooManyRedirects, + ValueError, + SSLError, + ) as e: + logger.warning( + f"Request failed for '{username}' on site '{validation_uri}': {e}" + ) continue except Exception as e: - logger.exception(f"Unexpected error while searching for username '{username}' on site '{uri}'") - task_status[username] = { - "status": "failed", - "error": str(e), - "found_sites": task_status[username].get("found_sites", []), - "checked_sites": task_status[username].get("checked_sites", 0), - "total_sites": task_status[username].get("total_sites", len(sites)), - } - raise + logger.exception(f"Unexpected error while searching for username \ + '{username}' on site '{validation_uri}'") + raise e - logger.info(f"Finished background search for username '{username}'. Found {len(sites_found)} matches.") - task_status[username]["status"] = "completed" + logger.info( + f"Finished background search for username '{username}'. Found {len(sites_found)} matches." + ) return sites_found -@app.get("/scan/{username}") -async def get_username_data(username: str, background_tasks: BackgroundTasks, refresh: str = "false"): - logger.info(f"Received scan request for '{username}' with refresh={refresh}") - - if username in task_status: - current_task_status = task_status[username]["status"] - if current_task_status in {"in_progress", "pending"}: - logger.info(f"Search already in progress for '{username}'") - return JSONResponse( - status_code=HTTPStatus.PROCESSING, - content={"message": f"Search for {username} in progress", "status": current_task_status}, - ) - - refresh = refresh.lower() - - if refresh == "false" and has_username_been_searched(username): - logger.info(f"Returning cached search results for '{username}'") - sites = get_sites_by_username(username) - return sites - - if refresh != "false": - logger.info(f"Invalidating cache for '{username}' and starting a fresh search") - delete_search_history(username) - background_tasks.add_task(search_for_username, username) - task_status[username] = {"status": "pending", "found_sites": [], "total_sites": 0, "checked_sites": 0} - logger.info(f"Scheduled background search task for '{username}'") - - data = {"message": f"Search for username {username} started."} - return JSONResponse(status_code=status.HTTP_202_ACCEPTED, content=data) - -@app.get("/scan/status/{username}") -async def get_search_status(username: str): - if username not in task_status: - raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Username search task not found.") - - return task_status[username] \ No newline at end of file +@app.get("/scan/{username}") +async def get_username_data(username: str): + """ + Endpoint to scan for the given username across multiple sites. It logs the incoming request and calls the search function to perform the scan, returning the results as a JSON response with a 200 OK status code. + """ + logger.info(f"Received scan request for '{username}'") + data = await search_for_username(username) + return data diff --git a/models.py b/models.py new file mode 100644 index 0000000..5e5cd4e --- /dev/null +++ b/models.py @@ -0,0 +1,26 @@ +from enum import Enum + +from pydantic import BaseModel, Field + + +class XUsernameAvailabilityReason(str, Enum): + taken = "taken" + available = "available" + invalid_username = "invalid_username" + + +class XValidationResponse(BaseModel): + valid: bool + reason: XUsernameAvailabilityReason + msg: str + desc: str + + +class SiteData(BaseModel): + title: str + profile_uri: str + validation_uri: str + + +class SiteResult(SiteData): + is_valid_profile: bool = Field(..., description="True when the profile exists") diff --git a/requirements.txt b/requirements.txt index f1417e4..a34a95b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,4 @@ fastapi[standard] httpx -websockets -anthropic python-dotenv +PyYAML diff --git a/sites.yaml b/sites.yaml new file mode 100644 index 0000000..5a4d5cb --- /dev/null +++ b/sites.yaml @@ -0,0 +1,4 @@ +sites: + - title: "X" + profile_uri: "https://x.com/{username}" + validation_uri: "https://api.x.com/i/users/username_available.json?username={username}"