Skip to content

Commit 5f46680

Browse files
taheeraahmedtaheeraahmed
committed
chg: add sync script for publications per researcher
1 parent bcb5f8e commit 5f46680

7 files changed

Lines changed: 164 additions & 24 deletions

File tree

‎src/datasync/nva/__init__.py‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,12 @@
22
# this allows to attach the scripts to the app
33
import typer
44

5-
from . import nva, nva_filter_resources, nva_search_resources
5+
from . import (
6+
nva,
7+
nva_filter_resources,
8+
nva_get_publications_per_researcher,
9+
nva_search_resources,
10+
)
611

712
app = typer.Typer(help="Commands to handle NVA tasks")
813

@@ -15,3 +20,7 @@
1520
app.command(help="Fetch and filter NVA data with flexible search parameters")(
1621
nva_search_resources.search_resources_api
1722
)
23+
24+
app.command(help="Fetch publications for researchers at NINA")(
25+
nva_get_publications_per_researcher.get_pubs_per_researcher
26+
)

‎src/datasync/nva/nva.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,6 @@ def run(
179179
bucket=bucket,
180180
prefix=prefix,
181181
credentials=credentials,
182-
region=region,
183182
)
184183

185184
log.info(

‎src/datasync/nva/nva_get_publications_per_researcher.py‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
# get nva id's https://s3-ext-1.nina.no/dms/nina/employees.parquet
2+
# https://api.nva.unit.no/search/resources?contributor=https://api.nva.unit.no/cristin/person/408357
3+
# use search_api function in nva_search_resources.py
4+
import typer
5+
6+
from ..settings import (
7+
log,
8+
)
9+
from .nva_search_resources import search_resources_api
10+
from .settings import (
11+
NINA_EMPLOYEES_PARQUET,
12+
NVA_S3_ACCESS_KEY,
13+
NVA_S3_BUCKET,
14+
NVA_S3_ENDPOINT_URL,
15+
NVA_S3_PREFIX,
16+
NVA_S3_REGION,
17+
NVA_S3_SECRET_KEY,
18+
)
19+
from .utils import (
20+
setup_duckdb_s3_connection,
21+
)
22+
23+
app = typer.Typer(help="Fetching publications from researchers at NINA")
24+
25+
26+
def _fetch_publications_for_researcher(
27+
nva_id: str,
28+
storage_s3_path: str,
29+
storage_access_key: str,
30+
storage_secret_key: str,
31+
storage_bucket: str,
32+
storage_prefix: str,
33+
storage_region: str,
34+
) -> str:
35+
log.debug("Fetching publications for researcher", nva_id=nva_id)
36+
search_resources_api(
37+
resource_name=f"publications_{nva_id}",
38+
filters=[f"contributor=https://api.nva.unit.no/cristin/person/{nva_id}"],
39+
apply_filter=False,
40+
storage_endpoint_url=storage_s3_path,
41+
storage_access_key=storage_access_key,
42+
storage_secret_key=storage_secret_key,
43+
storage_bucket=storage_bucket,
44+
storage_prefix=f"{storage_prefix}/{nva_id}/",
45+
storage_region=storage_region,
46+
)
47+
return nva_id
48+
49+
50+
@app.command()
51+
def get_pubs_per_researcher(
52+
employees_parquet: str = NINA_EMPLOYEES_PARQUET,
53+
storage_s3_path: str = NVA_S3_ENDPOINT_URL,
54+
storage_access_key: str = NVA_S3_ACCESS_KEY,
55+
storage_secret_key: str = NVA_S3_SECRET_KEY,
56+
storage_bucket: str = NVA_S3_BUCKET,
57+
storage_prefix: str = NVA_S3_PREFIX,
58+
storage_region: str = NVA_S3_REGION,
59+
):
60+
log.debug("Fetching ALL publications from researchers at NINA")
61+
62+
if not employees_parquet:
63+
log.error(
64+
"Path to employees parquet file is not set."
65+
"Set the NINA_EMPLOYEES_PARQUET environment variable."
66+
)
67+
raise typer.Exit(code=1)
68+
69+
con = setup_duckdb_s3_connection(
70+
endpoint_url=storage_s3_path,
71+
access_key=storage_access_key,
72+
secret_key=storage_secret_key,
73+
region=storage_region,
74+
)
75+
76+
employees = con.read_parquet(employees_parquet).to_df()
77+
78+
# get all employees with missing nva_id and log their names for debugging
79+
missing_nva_id = employees["nva_id"].isna().sum()
80+
if missing_nva_id > 0:
81+
missing_names = (
82+
employees[employees["nva_id"].isna()]["firstname"]
83+
+ " "
84+
+ employees[employees["nva_id"].isna()]["lastname"]
85+
).tolist()
86+
log.warning(
87+
"Employees missing NVA IDs",
88+
missing_count=missing_nva_id,
89+
total_employees=len(employees),
90+
missing_names=missing_names,
91+
)
92+
con.close()
93+
nva_ids = [str(nva_id) for nva_id in employees["nva_id"].dropna().unique()]
94+
95+
failed = []
96+
for nva_id in nva_ids:
97+
try:
98+
_fetch_publications_for_researcher(
99+
nva_id,
100+
storage_s3_path,
101+
storage_access_key,
102+
storage_secret_key,
103+
storage_bucket,
104+
storage_prefix,
105+
storage_region,
106+
)
107+
except Exception as e:
108+
log.error(
109+
"Failed to fetch publications for researcher",
110+
nva_id=nva_id,
111+
error=str(e),
112+
)
113+
failed.append(nva_id)
114+
115+
if failed:
116+
log.warning(
117+
"Some researchers failed to fetch", failed=failed, count=len(failed)
118+
)
119+
120+
log.info(
121+
"Fetched all publications for researchers",
122+
total_researchers=len(nva_ids),
123+
failed=len(failed),
124+
succeeded=len(nva_ids) - len(failed),
125+
)

‎src/datasync/nva/nva_search_resources.py‎

Lines changed: 10 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ def search_resources_api(
126126
log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
127127
raise typer.Exit(code=1)
128128

129-
log.info(f"Fetching NVA data with search parameters: {search_params}")
129+
log.debug(f"Fetching NVA data with search parameters: {search_params}")
130130

131131
credentials = create_s3_credentials(
132132
endpoint_url=storage_endpoint_url,
@@ -140,10 +140,9 @@ def search_resources_api(
140140
bucket=storage_bucket,
141141
prefix=storage_prefix,
142142
credentials=credentials,
143-
region=storage_region,
144143
)
145144

146-
log.info(f"Fetching resources from {base_url}")
145+
log.debug(f"Fetching resources from {base_url}")
147146
load_info = pipeline.run(
148147
nva_search_source(
149148
base_url=base_url,
@@ -154,16 +153,15 @@ def search_resources_api(
154153
loader_file_format="parquet",
155154
)
156155

157-
log.info("Pipeline run completed", load_info=load_info)
156+
log.debug("Pipeline run completed", load_info=load_info)
158157

158+
con = setup_duckdb_s3_connection(
159+
endpoint_url=storage_endpoint_url,
160+
access_key=storage_access_key,
161+
secret_key=storage_secret_key,
162+
region=storage_region,
163+
)
159164
if apply_filter:
160-
con = setup_duckdb_s3_connection(
161-
endpoint_url=storage_endpoint_url,
162-
access_key=storage_access_key,
163-
secret_key=storage_secret_key,
164-
region=storage_region,
165-
)
166-
167165
apply_filter_transformation(
168166
con=con,
169167
bucket=storage_bucket,
@@ -175,7 +173,7 @@ def search_resources_api(
175173
write_timestamp(con, storage_bucket, storage_prefix)
176174
con.close()
177175

178-
log.info("NVA data sync completed")
176+
log.debug("NVA data sync completed")
179177
log.info(
180178
f"Data available at: {storage_endpoint_url}/{storage_bucket}/{storage_prefix}/"
181179
f"main/{resource_name}.parquet"

‎src/datasync/nva/settings.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,9 @@
1313
NVA_S3_PREFIX = env("NVA_S3_PREFIX", default="")
1414
NVA_S3_REGION = env("NVA_S3_REGION", default="us-east-1")
1515

16+
# Employees parquet file path
17+
NINA_EMPLOYEES_PARQUET = env("NINA_EMPLOYEES_PARQUET", default="")
18+
1619
# Valid search parameters accepted by the NVA API search endpoint
1720
# Based on: https://swagger-ui.nva.unit.no/#/NVA%20Public%20Search%20API/searchResources
1821
VALID_PARAMS_NVA_API = {

‎src/datasync/nva/utils.py‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ def get_nva_resources(
2727
Yields:
2828
Resources from the NVA API
2929
"""
30-
log.info(f"Fetching resources with params: {search_params}")
30+
log.debug(f"Fetching resources with params: {search_params}")
3131
yield from client.paginate(
3232
"search/resources",
3333
method="GET",
@@ -103,19 +103,16 @@ def create_pipeline(
103103
bucket: str,
104104
prefix: str,
105105
credentials: AwsCredentials,
106-
region: str,
107106
) -> dlt.Pipeline:
108107
"""Create a DLT pipeline with filesystem destination."""
109108
return dlt.pipeline(
110109
pipeline_name=pipeline_name,
111110
destination=filesystem(
112-
region_name=region,
113-
bucket_url=f"s3://{bucket}/{prefix}",
111+
bucket_url=f"s3://{bucket}/{prefix.rstrip('/')}",
114112
credentials=credentials,
115113
layout="{table_name}.{ext}",
116114
),
117115
dataset_name="main",
118-
progress="log",
119116
)
120117

121118

@@ -154,6 +151,7 @@ def write_timestamp(
154151
) -> str:
155152
"""Write last successful run timestamp to S3."""
156153
timestamp = datetime.datetime.now().isoformat()
154+
prefix = prefix.rstrip("/")
157155
con.execute(f"""
158156
COPY (SELECT '{timestamp}' as last_successful_run)
159157
TO 's3://{bucket}/{prefix}/last_successful_run.parquet'

‎uv.lock‎

Lines changed: 13 additions & 5 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)