Spaces:
Sleeping
Sleeping
File size: 3,749 Bytes
35b3f62 9426cdd 35b3f62 fcd14e1 558c227 35b3f62 d210108 35b3f62 d210108 a87af2b 35b3f62 558c227 35b3f62 558c227 35b3f62 602a998 35b3f62 |
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 |
import json
import os
import requests
import pandas as pd
from pathlib import Path
from .date_verifier import is_after_start
def google_search(query, api_key, search_engine_id, start_date, end_date):
# print(f"[SYSTEM] Calling Google Search API for: {query}")
sort = f"date:r:{start_date}:{end_date}" #20241230:20250130
url = "https://www.googleapis.com/customsearch/v1"
params = {
"q": query,
"key": api_key,
"cx": search_engine_id,
"num": 10,
"sort": sort,
"cr": "countryUK",
"gl": "uk"
}
try:
response = requests.get(url, params=params)
response.raise_for_status()
return response.json().get("items", [])
except Exception as e:
print(f"[ERROR] Google Search Failed: {e}")
return []
def save_tsv(file_name, id_value, string_value, value_list, query):
data = {
'ID': id_value,
'String': string_value,
'ListValue': value_list,
'query': query
}
df = pd.DataFrame(data)
df.to_csv(file_name, sep='\t', index=False, header=False)
def ensure_directory_exists(path):
dir_path = Path(path).expanduser().resolve().parent
if not str(dir_path).startswith("/home") and not str(dir_path).startswith("/data") and not str(dir_path).startswith("outputs"):
raise ValueError(f"[ERROR] Unsafe path: {dir_path}")
dir_path.mkdir(parents=True, exist_ok=True)
def run_augmented_searching(qa_file, pipeline_base_dir, suggestion_meta, pledge_author, pledge_date, start_date, end_date):
if suggestion_meta==None:
qa_lines = open(f"{qa_file}","r").read()
qa_lines = json.loads(qa_lines)
claim_text = f"{pledge_author}: {qa_lines['claim']} ({pledge_date})"
idx=0
else:
# claim_text = suggestion_meta["text"]
idx = suggestion_meta["index"]
qa_lines = open(f"{qa_file}","r").readlines()[idx]
qa_lines = json.loads(qa_lines)
claim_text = f"{qa_lines['claim']}"
api_key = os.environ.get("GOOGLE_API_KEY")
search_engine_id = os.environ.get("GOOGLE_SEARCH_CX")
if not api_key or not search_engine_id:
raise EnvironmentError("[ERROR] GOOGLE_API_KEY and GOOGLE_SEARCH_CX must be set in environment.")
# base_dir = pipeline_base_dir
tsv_file_path = os.path.join(pipeline_base_dir, "augmented_search_results.tsv")
ensure_directory_exists(tsv_file_path)
urls = []
string_values = []
queries = []
questions = []
questions = [evidence["question"] for evidence in qa_lines["evidence"] if evidence["question"] not in questions]
questions = questions[:10]
results = google_search(claim_text, api_key, search_engine_id, start_date, end_date)
for result in results:
if result["link"] not in urls and "fullfact.org/government-tracker" not in result["link"] and is_after_start(result["link"], start_date):
string_values.append("claim")
urls.append(result["link"])
queries.append(f"{pledge_author}: {claim_text}")
for question in questions:
results = google_search(f"{question}", api_key, search_engine_id, start_date, end_date)
for result in results:
if result["link"] not in urls and "fullfact.org/government-tracker" not in result["link"] and is_after_start(result["link"], start_date):
string_values.append("question")
urls.append(result["link"])
queries.append(f"{question}")
urls = list(dict.fromkeys(urls))
save_tsv(str(tsv_file_path), [0] * len(urls), string_values, urls, queries)
print(f"[SYSTEM] Saved {len(urls)} URLs for claim {idx} to {tsv_file_path}")
return str(tsv_file_path)
|