Skip to content

Commit 19e3a30

Browse files
taheeraahmedtaheeraahmed
committed
chg: add sync script for publications per researcher
1 parent 12ba31d commit 19e3a30

6 files changed

Lines changed: 321 additions & 22 deletions

File tree

‎pyproject.toml‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,8 @@ dependencies = [
4444
"pyyaml>=6.0.3",
4545
"deepdiff>=8.6.1",
4646
"xmldiff>=2.7.0",
47-
"xmlschema>=4.3.1"
47+
"xmlschema>=4.3.1",
48+
"pandas>=2.3.3"
4849
]
4950
description = ""
5051
license = "GPL-3.0+"
@@ -63,7 +64,7 @@ ubw_sync = "datasync.ubw:app"
6364
known_first_party = ["src"]
6465

6566
[tool.deptry.per_rule_ignores]
66-
DEP002 = ["rich", "numpy", "setuptools"]
67+
DEP002 = ["rich", "numpy", "setuptools", "pandas"]
6768

6869
[tool.hatch.metadata]
6970
allow-direct-references = true

‎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: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
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+
) -> str:
24+
log.debug("Fetching publications for researcher", nva_id=nva_id)
25+
search_resources_api(
26+
resource_name=f"publications_{nva_id}",
27+
filters=[f"contributor=https://api.nva.unit.no/cristin/person/{nva_id}"],
28+
apply_filter=False,
29+
storage_endpoint_url=storage_s3_path,
30+
storage_access_key=storage_access_key,
31+
storage_secret_key=storage_secret_key,
32+
storage_bucket=storage_bucket,
33+
storage_prefix=f"{storage_prefix}/{nva_id}/",
34+
storage_region=storage_region,
35+
add_timestamp=False,
36+
)
37+
return nva_id
38+
39+
40+
@app.command()
41+
def get_pubs_per_researcher(
42+
employees_parquet: str = typer.Option(
43+
...,
44+
envvar="NINA_EMPLOYEES_PARQUET",
45+
help="Path to employees parquet file",
46+
),
47+
storage_s3_path: str = typer.Option(
48+
...,
49+
envvar="NVA_S3_ENDPOINT_URL",
50+
help="S3 endpoint URL for storage",
51+
),
52+
storage_access_key: str = typer.Option(
53+
..., envvar="NVA_S3_ACCESS_KEY", help="S3 access key for storage"
54+
),
55+
storage_secret_key: str = typer.Option(
56+
..., envvar="NVA_S3_SECRET_KEY", help="S3 secret key for storage"
57+
),
58+
storage_bucket: str = typer.Option(
59+
..., envvar="NVA_S3_BUCKET", help="S3 bucket for storage"
60+
),
61+
storage_prefix: str = typer.Option(
62+
..., envvar="NVA_S3_PREFIX", help="S3 prefix for storage"
63+
),
64+
storage_region: str = typer.Option(
65+
"us-east-1", envvar="NVA_S3_REGION", help="S3 region for storage"
66+
),
67+
):
68+
log.debug("Fetching ALL publications from researchers at NINA")
69+
70+
con = setup_duckdb_s3_connection(
71+
endpoint_url=storage_s3_path,
72+
access_key=storage_access_key,
73+
secret_key=storage_secret_key,
74+
region=storage_region,
75+
)
76+
77+
employees = con.read_parquet(employees_parquet).to_df()
78+
79+
# get all employees with missing nva_id and log their names for debugging
80+
missing_nva_id = employees["nva_id"].isna().sum()
81+
if missing_nva_id > 0:
82+
missing_names = (
83+
employees[employees["nva_id"].isna()]["firstname"]
84+
+ " "
85+
+ employees[employees["nva_id"].isna()]["lastname"]
86+
).tolist()
87+
log.warning(
88+
"Employees missing NVA IDs",
89+
missing_count=missing_nva_id,
90+
total_employees=len(employees),
91+
missing_names=missing_names,
92+
)
93+
con.close()
94+
nva_ids = [str(nva_id) for nva_id in employees["nva_id"].dropna().unique()]
95+
96+
failed = []
97+
for nva_id in nva_ids:
98+
try:
99+
_fetch_publications_for_researcher(
100+
nva_id,
101+
storage_s3_path,
102+
storage_access_key,
103+
storage_secret_key,
104+
storage_bucket,
105+
storage_prefix,
106+
storage_region,
107+
)
108+
except Exception as e:
109+
log.error(
110+
"Failed to fetch publications for researcher",
111+
nva_id=nva_id,
112+
error=str(e),
113+
)
114+
failed.append(nva_id)
115+
116+
if failed:
117+
log.warning(
118+
"Some researchers failed to fetch", failed=failed, count=len(failed)
119+
)
120+
121+
log.info(
122+
"Fetched all publications for researchers",
123+
total_researchers=len(nva_ids),
124+
failed=len(failed),
125+
succeeded=len(nva_ids) - len(failed),
126+
)
127+
128+
con = setup_duckdb_s3_connection(
129+
endpoint_url=storage_s3_path,
130+
access_key=storage_access_key,
131+
secret_key=storage_secret_key,
132+
region=storage_region,
133+
)
134+
write_timestamp(con, storage_bucket, storage_prefix)
135+
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'

0 commit comments

Comments
 (0)