Spaces:
Running
Running
File size: 4,472 Bytes
6523110 dfef9ae 6523110 |
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 126 127 128 129 130 131 |
from urllib.parse import urlparse
import logging
log = logging.getLogger(__name__)
from time import sleep
import requests
helpers_session = requests.Session()
from wikibaseintegrator.wbi_helpers import get_user_agent
import pandas as pd
from string import Template
queries = False
wikibase_api_url = 'https://fashionwiki.wikibase.cloud/w/api.php'
config = {
"SPARQL_ENDPOINT_URL": "https://fashionwiki.wikibase.cloud/query/sparql",
'USER_AGENT': 'YourBotName/1.0 (https://yourwebsite.org/bot-info)',
'WIKIBASE_URL': wikibase_api_url,
}
wikibase_properties_id = {'instance of': 'P2',
'reference URL': 'P24',
'start time': 'P15',
'end time': 'P16',
'occupation title': 'P25',
'educated at': 'P9',
'employer': 'P10',
'work location': 'P7',
'award received': 'P18',
'point in time': 'P28',
'exact match': 'P23',
'date of birth': 'P3',
'place of birth': 'P4',
'date of death': 'P5',
'country of citizenship': 'P6',
'occupation': 'P19',
'sex or gender': 'P8',
'official website': 'P17',
'perfumes': 'P27',
'who wears it': 'P26',
'inception': 'P11',
'headquarters location': 'P12',
'parent organization': 'P13',
'founded by': 'P14',
'owned by': 'P22',
'industry': 'P20',
'country': 'P30',
'total revenue': 'P21',
'designer employed': 'P29',
'country of origin': 'P30',
'fashion collection': 'P31',
'fashion season': 'P32',
'fashion show location': 'P33',
'description of fashion collection': 'P34',
'image of fashion collection': 'P35',
'editor of fashion collection description': 'P36',
'date of fashion collection': 'P37',
'fashion show category': 'P38',
'fashion house X fashion collection': 'P39'}
classes_wikibase = {'fashion designer': 'Q5',
'fashion house': 'Q1',
'business': 'Q9',
'academic institution': 'Q2',
'geographic location': 'Q4',
'fashion award': 'Q8',
'gender': 'Q6',
'occupation': 'Q7',
'human': 'Q36',
'organization': 'Q3',
'brand': 'Q38',
'lifestyle brand': 'Q3417',
'privately held company': 'Q1729',
'fashion season': 'Q8199',
'fashion show category': 'Q8200',
'fashion season collection': 'Q8201',
'fashion journalist': 'Q8207'}
def execute_sparql_query(query: str, prefix: str | None = None, endpoint: str | None = None, user_agent: str | None = None, max_retries: int = 1000, retry_after: int = 60) -> dict:
"""
Execute any SPARQL query with the provided parameters.
"""
sparql_endpoint_url = str(endpoint or config['SPARQL_ENDPOINT_URL'])
user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)
hostname = urlparse(sparql_endpoint_url).hostname
if hostname and hostname.endswith(('wikidata.org', 'wikipedia.org', 'wikimedia.org')) and user_agent is None:
log.warning('WARNING: Please set a user agent if you interact with a Wikimedia Foundation instance.')
if prefix:
query = prefix + '\n' + query
headers = {
'Accept': 'application/sparql-results+json',
'User-Agent': get_user_agent(user_agent),
'Content-Type': 'application/sparql-query' # Correct Content-Type
}
# Attempt to make the request
for _ in range(max_retries):
try:
# Use 'data' instead of 'params' for the POST request to SPARQL
response = helpers_session.post(sparql_endpoint_url, data=query, headers=headers)
except requests.exceptions.ConnectionError as e:
log.exception("Connection error: %s. Sleeping for %d seconds.", e, retry_after)
sleep(retry_after)
continue
if response.status_code in (500, 502, 503, 504):
log.error("Service unavailable (HTTP Code %d). Sleeping for %d seconds.", response.status_code, retry_after)
sleep(retry_after)
continue
if response.status_code == 429:
if 'retry-after' in response.headers:
retry_after = int(response.headers['retry-after'])
log.error("Too Many Requests (429). Sleeping for %d seconds", retry_after)
sleep(retry_after)
continue
response.raise_for_status() # Raise any non-success status code
return response.json() # Return the JSON result if successful
raise Exception(f"No result after {max_retries} retries.")
def get_results_to_df( query):
results = execute_sparql_query(query)
df = pd.DataFrame(results["results"]["bindings"])
df = df.map(lambda x: x['value'] if pd.notnull(x) else None)
return df
|