Skip to content

Commit db5b1ad

Browse files
taheeraahmedtaheeraahmed
committed
chg(nva): add sync script for publications per researcher
1 parent 12ba31d commit db5b1ad

5 files changed

Lines changed: 196 additions & 20 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_get_publications_per_researcher.py‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
import typer
2+
3+
from ..settings import (
4+
log,
5+
)
6+
from .nva_search_resources import search_resources_api
7+
from .utils import (
8+
setup_duckdb_s3_connection,
9+
write_timestamp,
10+
)
11+
12+
app = typer.Typer(help="Fetching publications from researchers at NINA")
13+
14+
15+
def _fetch_publications_for_researcher(
16+
nva_id: str,
17+
storage_s3_path: str,
18+
storage_access_key: str,
19+
storage_secret_key: str,
20+
storage_bucket: str,
21+
storage_prefix: str,
22+
storage_region: str,
23+
base_url: str = "https://api.nva.unit.no/",
24+
) -> str:
25+
log.debug("Fetching publications for researcher", nva_id=nva_id)
26+
search_resources_api(
27+
resource_name=f"publications_{nva_id}",
28+
base_url=base_url,
29+
filters=[f"contributor=https://api.nva.unit.no/cristin/person/{nva_id}"],
30+
apply_filter=False,
31+
storage_endpoint_url=storage_s3_path,
32+
storage_access_key=storage_access_key,
33+
storage_secret_key=storage_secret_key,
34+
storage_bucket=storage_bucket,
35+
storage_prefix=f"{storage_prefix}/{nva_id}/",
36+
storage_region=storage_region,
37+
add_timestamp=False,
38+
)
39+
return nva_id
40+
41+
42+
@app.command()
43+
def get_pubs_per_researcher(
44+
employees_parquet: str = typer.Option(
45+
...,
46+
envvar="NINA_EMPLOYEES_PARQUET",
47+
help="Path to employees parquet file",
48+
),
49+
storage_s3_path: str = typer.Option(
50+
...,
51+
envvar="NVA_S3_ENDPOINT_URL",
52+
help="S3 endpoint URL for storage",
53+
),
54+
storage_access_key: str = typer.Option(
55+
..., envvar="NVA_S3_ACCESS_KEY", help="S3 access key for storage"
56+
),
57+
storage_secret_key: str = typer.Option(
58+
..., envvar="NVA_S3_SECRET_KEY", help="S3 secret key for storage"
59+
),
60+
storage_bucket: str = typer.Option(
61+
..., envvar="NVA_S3_BUCKET", help="S3 bucket for storage"
62+
),
63+
storage_prefix: str = typer.Option(
64+
..., envvar="NVA_S3_PREFIX", help="S3 prefix for storage"
65+
),
66+
storage_region: str = typer.Option(
67+
"us-east-1", envvar="NVA_S3_REGION", help="S3 region for storage"
68+
),
69+
test: bool = typer.Option(
70+
False,
71+
help="If set, will only fetch publications for the first 3 researchers",
72+
),
73+
):
74+
log.debug("Fetching ALL publications from researchers at NINA")
75+
76+
con = setup_duckdb_s3_connection(
77+
endpoint_url=storage_s3_path,
78+
access_key=storage_access_key,
79+
secret_key=storage_secret_key,
80+
region=storage_region,
81+
)
82+
83+
employees = con.read_parquet(employees_parquet) # noqa: F841
84+
85+
# get all employees with missing nva_id and log their names for debugging
86+
summary = con.execute("""
87+
SELECT
88+
COUNT(*) FILTER (WHERE nva_id IS NULL) AS missing_count,
89+
COUNT(*) AS total,
90+
array_agg(firstname || ' ' || lastname) FILTER (WHERE nva_id IS NULL) AS missing_names,
91+
array_agg(DISTINCT nva_id::VARCHAR) FILTER (WHERE nva_id IS NOT NULL) AS nva_ids
92+
FROM employees
93+
""").fetchone() # noqa: E501
94+
missing_count, total, missing_names, nva_ids = summary
95+
96+
if missing_count > 0:
97+
log.warning(
98+
"Employees missing NVA IDs",
99+
missing_count=missing_count,
100+
total_employees=total,
101+
missing_names=missing_names,
102+
)
103+
con.close()
104+
105+
if test:
106+
nva_ids = nva_ids[:3]
107+
log.warning(
108+
"Test mode enabled, only fetching publications for first 3 researchers",
109+
nva_ids=nva_ids,
110+
)
111+
112+
failed = []
113+
for nva_id in nva_ids:
114+
try:
115+
_fetch_publications_for_researcher(
116+
nva_id,
117+
storage_s3_path,
118+
storage_access_key,
119+
storage_secret_key,
120+
storage_bucket,
121+
storage_prefix,
122+
storage_region,
123+
)
124+
except Exception as e:
125+
log.error(
126+
"Failed to fetch publications for researcher",
127+
nva_id=nva_id,
128+
error=str(e),
129+
)
130+
failed.append(nva_id)
131+
132+
if failed:
133+
log.warning(
134+
"Some researchers failed to fetch", failed=failed, count=len(failed)
135+
)
136+
137+
log.info(
138+
"Fetched all publications for researchers",
139+
total_researchers=len(nva_ids),
140+
failed=len(failed),
141+
succeeded=len(nva_ids) - len(failed),
142+
)
143+
144+
con = setup_duckdb_s3_connection(
145+
endpoint_url=storage_s3_path,
146+
access_key=storage_access_key,
147+
secret_key=storage_secret_key,
148+
region=storage_region,
149+
)
150+
write_timestamp(con, storage_bucket, storage_prefix)
151+
con.close()

‎src/datasync/nva/nva_search_resources.py‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ def search_resources_api(
5959
storage_region: str = typer.Option(
6060
"us-east-1", "--storage-region", help="S3 region for storage"
6161
),
62+
add_timestamp: bool = typer.Option(
63+
True, help="Whether to write last successful run timestamp to S3"
64+
),
6265
):
6366
"""
6467
Get resources from NVA API
@@ -128,7 +131,7 @@ def search_resources_api(
128131
log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
129132
raise typer.Exit(code=1)
130133

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

133136
credentials = create_s3_credentials(
134137
endpoint_url=storage_endpoint_url,
@@ -145,8 +148,8 @@ def search_resources_api(
145148
region=storage_region,
146149
)
147150

148-
log.info(f"Fetching resources from {base_url}")
149-
pipeline.run(
151+
log.debug(f"Fetching resources from {base_url}")
152+
load_info = pipeline.run(
150153
nva_search_source(
151154
base_url=base_url,
152155
resource_name=resource_name,
@@ -156,14 +159,15 @@ def search_resources_api(
156159
loader_file_format="parquet",
157160
)
158161

159-
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-
)
162+
log.debug("Pipeline run completed", load_info=load_info)
166163

164+
con = setup_duckdb_s3_connection(
165+
endpoint_url=storage_endpoint_url,
166+
access_key=storage_access_key,
167+
secret_key=storage_secret_key,
168+
region=storage_region,
169+
)
170+
if apply_filter:
167171
apply_filter_transformation(
168172
con=con,
169173
bucket=storage_bucket,
@@ -172,11 +176,14 @@ def search_resources_api(
172176
check_additional_identifiers=True,
173177
)
174178

175-
write_timestamp(con, storage_bucket, storage_prefix)
179+
if add_timestamp:
180+
write_timestamp(con, storage_bucket, storage_prefix)
176181
con.close()
177182

183+
log.debug("NVA data sync completed")
178184
log.info(
179-
f"Data available at: {storage_endpoint_url}/{storage_bucket}/{storage_prefix}/"
185+
f"Data available at: {storage_endpoint_url}/{storage_bucket}/"
186+
f"{storage_prefix.rstrip('/')}/"
180187
f"main/{resource_name}.parquet"
181188
)
182189

‎src/datasync/nva/utils.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ def get_nva_resources(
2626
Yields:
2727
Resources from the NVA API
2828
"""
29-
log.info(f"Fetching resources with params: {search_params}")
29+
log.debug(f"Fetching resources with params: {search_params}")
3030
yield from client.paginate(
3131
"search/resources",
3232
method="GET",
@@ -106,7 +106,7 @@ def create_pipeline(
106106
pipeline_name=pipeline_name,
107107
destination=filesystem(
108108
region_name=region,
109-
bucket_url=f"s3://{bucket}/{prefix}",
109+
bucket_url=f"s3://{bucket}/{prefix.rstrip('/')}",
110110
credentials=credentials,
111111
layout="{table_name}.{ext}",
112112
),
@@ -150,6 +150,7 @@ def write_timestamp(
150150
) -> str:
151151
"""Write last successful run timestamp to S3."""
152152
timestamp = datetime.datetime.now().isoformat()
153+
prefix = prefix.rstrip("/")
153154
con.execute(f"""
154155
COPY (SELECT '{timestamp}' as last_successful_run)
155156
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)