From 52b1e86d6b73a0a0e2a949bc9453b22df53e1329 Mon Sep 17 00:00:00 2001 From: mgrani Date: Fri, 2 May 2025 12:47:55 +0200 Subject: [PATCH] feat: query sites command now allows to filter for a list of urls provided in a file. --- Readme.md | 2 +- owilix/cmd/query.py | 157 ++++++++++++++++++++++++++++++++++++-------- pyproject.toml | 1 + tests/data/urls.csv | 5 ++ 4 files changed, 135 insertions(+), 30 deletions(-) create mode 100644 tests/data/urls.csv diff --git a/Readme.md b/Readme.md index 1fb8c2f..1a53ae5 100644 --- a/Readme.md +++ b/Readme.md @@ -29,7 +29,7 @@ conda create -n owi pip python=3.11 conda activate owi # Install required packages pip install py4lexis --index-url https://opencode.it4i.eu/api/v4/projects/107/packages/pypi/simple -pip install owilix --index-url https://opencode.it4i.eu/api/v4/projects/92/packages/pypi/simple +pip install owilix --index-url https://opencode.it4i.eu/api/v4/projects/92/packages/pypi/simple` # Verify installation owilix --help ``` diff --git a/owilix/cmd/query.py b/owilix/cmd/query.py index 6cf0612..2923040 100644 --- a/owilix/cmd/query.py +++ b/owilix/cmd/query.py @@ -13,6 +13,40 @@ from owilix.core.duckdb import OWIlixSQLQuery, OWIDuckDBCopyExecutor, OWIDuckDBS from owilix.core.metadata import Dataset, fill_metadata from owilix.core.stream import OWIDuckDBArrow from owilix.core.utils import OwilixJSONEncoder +from url_normalize import url_normalize +from urllib.parse import urlparse + + +def extract_domain_components(url): + try: + normalized = url_normalize(url) + parsed = urlparse(normalized) + netloc = parsed.hostname or '' + parts = netloc.split('.') + scheme = parsed.scheme + path = parsed.path + if len(parts) < 2: + return { + 'url_scheme': scheme, + 'url_subdomain': '', + 'url_domain': netloc, + 'url_suffix': '', + 'url_path': path, + } + return { + 'url_scheme': scheme, + 'url_subdomain': '.'.join(parts[:-2]) if len(parts) > 2 else '', + 'url_domain': parts[-2], + 'url_suffix': parts[-1], + 'url_path': path, + } + except Exception: + return None + + + + + class QueryCommands(BaseCommand): @@ -92,6 +126,30 @@ class QueryCommands(BaseCommand): return dict(consolidated_data) + def _process_query_results(self, db, all_files, as_json, page_size=10): + show, count = [], 0 + _files_processed = [] + for results in db.query_aggregator(): + if not results.success: + self.console.print(f"[red]Error when running query: [/red]" + str(results.error)) + continue + if not as_json: + _files_processed.extend([f[0] for f in results.parquet_batch.files]) + show += results.rows + while len(show) > page_size: + self.console.print(f"Showing results {count} to {count + page_size} (file progress: {len(set(_files_processed)) / len(set(all_files)) * 100}%") + self.show_detail_table(show[:page_size]) + show, count = show[page_size:], count + page_size + if not ask_yes_no(self.console, f"Continue"): + return + else: + for i in results.rows: + try: + print(json.dumps(i, cls=OwilixJSONEncoder)) + except Exception as e: + self.console.print_exception() + self.console.log(f"Error when parsing json") + @QueryCommands.register def less( self, local_specifier: str, remote_specifier: str, select: str = "url,title,plain_text", @@ -122,10 +180,10 @@ def less( self, local_specifier: str, remote_specifier: str, page_size (int): Number of elements to be displayed per page. Defaults to 10. as_json (bool): If True, prints a JSON formated string for every document instead of printing it to the console in a formatted way. Defaults to False. """ - all_files = self.group_all_files_by_fs(self.get_all_files(files, local_specifier, remote_specifier,print_it=not as_json)) + all_files = self.group_all_files_by_fs(self.get_all_files(files, local_specifier, remote_specifier, print_it=not as_json)) if not as_json: - self.console.print(f"Found '{sum([len(v) for v in all_files.values()])}' parquet files " - f"in {len(all_files.keys())} filesystems. Running queries against them.") + self.console.print(f"Found '{sum([len(v) for v in all_files.values()])}' parquet files in {len(all_files)} filesystems. Running queries.") + sql = (OWIlixSQLQuery.from_templates("pq_select") .select(select) .where(where) @@ -135,39 +193,80 @@ def less( self, local_specifier: str, remote_specifier: str, .limit(limit)) db = OWIDuckDBSelectExecutor(all_files, sql, - pq_batch_size=pq_batch_size, - batch_size=batch_size, - prefetch=prefetch) - show, count = [], 0 - _files_processed = [] - for results in db.query_aggregator(): - if not results.success: - self.console.print(f"[red]Error when running query: [/red]" + str(results.error) ) - continue - if not as_json: - _files_processed.extend([f[0] for f in results.parquet_batch.files]) - show = show + results.rows - while len(show)>page_size: - self.console.print(f"Showing results {count} to {count+page_size} (file progress: {len(set(_files_processed))/len(set(all_files))*100}%") - self.show_detail_table(show[:page_size]) - show, count = show[page_size:], count+page_size - if not ask_yes_no(self.console, f"Continue"): - return - else: - for i in results.rows: - try: - print(json.dumps(i, cls=OwilixJSONEncoder)) - except Exception as e: - self.console.print_exception() - self.console.log(f"Error when parsing json") + pq_batch_size=pq_batch_size, + batch_size=batch_size, + prefetch=prefetch) + + self._process_query_results(db, all_files, as_json, page_size) + return CommandResult(success=True, object=all_files, msg=f"Shown {len(all_files)}") + +@QueryCommands.register +def sites(self, local_specifier: str, remote_specifier: str, urls_file: str, + select: str = "url,url_subdomain,url_domain,url_suffix", + limit: Optional[int] = None, files: str = "**/*.parquet", + as_json: bool = False, pq_batch_size: int = 1, batch_size: int = 100, + prefetch: int = 1, page_size: int = 10): + """ + Executes a query that filters records to only include rows from the provided list of URLs or sites. + The list of sites should be passed via a file. Each site is normalized and parsed into its subdomain, + domain, and suffix, and a WHERE clause is constructed accordingly. + Args: + local_specifier (str): Specifier to filter datasets locally. + remote_specifier (str): Specifier to filter datasets remotely. + urls_file (str): Path to a file containing URLs (one per line). + select (str): Columns to select. Defaults to "url,url_subdomain,url_domain,url_suffix". + limit (int, optional): Maximum number of results to return. + files (str): Glob pattern for selecting Parquet files. Defaults to "**/*.parquet". + as_json (bool): Whether to print the results in JSON format. Defaults to False. + pq_batch_size (int): Parquet file batch size. Defaults to 1. + batch_size (int): Row batch size. Defaults to 100. + prefetch (int): Number of batches to prefetch. Defaults to 1. + page_size (int): Rows per page for interactive display. Defaults to 10. + """ + with open(urls_file, 'r') as f: + urls = list(set(line.strip() for line in f if line.strip())) + filters = [extract_domain_components(url) for url in urls] + filters = [f for f in filters if f is not None] + filters = [dict(t) for t in {tuple(f.items()) for f in filters}] # deduplicate dicts + if not as_json: + self.console.print(f"Found {len(filters)} unique sites to query: {filters}") + + clauses = [] + for f in filters: + clause = f"(url_domain = '{f['url_domain']}' AND url_suffix = '{f['url_suffix']}'" + if f['url_subdomain']: + clause += f" AND url_subdomain = '{f['url_subdomain']}'" + if f['url_scheme'] and False: + clause += f" AND url_scheme = '{f['url_scheme']}'" + if f['url_path'] and f['url_path']!='/' and f['url_path']!='' : + clause += f" AND url_path = '{f['url_path']}'" + clause += ")" + clauses.append(clause) + + where_clause = " OR ".join(clauses) + + all_files = self.group_all_files_by_fs(self.get_all_files(files, local_specifier, remote_specifier, print_it=not as_json)) + if not as_json: + self.console.print(f"Found '{sum([len(v) for v in all_files.values()])}' parquet files in {len(all_files)} filesystems. Running site query.") + sql = (OWIlixSQLQuery.from_templates("pq_select") + .select(select) + .where(where_clause) + .groupby("") + .partitioned_by("") + .postfix("") + .limit(limit)) + db = OWIDuckDBSelectExecutor(all_files, sql, + pq_batch_size=pq_batch_size, + batch_size=batch_size, + prefetch=prefetch) + self._process_query_results(db, all_files, as_json, page_size) return CommandResult(success=True, object=all_files, msg=f"Shown {len(all_files)}") - @QueryCommands.register def aggregate( self, local_specifier: str, remote_specifier: str, select: str = "url,title,plain_text", diff --git a/pyproject.toml b/pyproject.toml index 3982473..648dbac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,7 @@ requests = "^2.32.3" python-dateutil = "^2.9.0.post0" #py4lexis = { git = "https://opencode.it4i.eu/lexis-platform/clients/py4lexis.git", branch = "develop" } # devleopment version of Lexis s3fs = "^2024.12.0" +url-normalize = "^2.2.1" [[tool.poetry.source]] diff --git a/tests/data/urls.csv b/tests/data/urls.csv new file mode 100644 index 0000000..9a91db9 --- /dev/null +++ b/tests/data/urls.csv @@ -0,0 +1,5 @@ +imdb.com +www.imdb.com +mtu.edu +www.wikitree.com +helsinki.fi \ No newline at end of file -- 2.51.2