Coverage for src/software_citation_sync/software_heritage.py: 89%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <github@a.arcangelomassari.com> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import time 

8from collections.abc import Mapping 

9from typing import cast 

10from urllib.parse import quote 

11 

12import requests 

13 

14SAVE_API = "https://archive.softwareheritage.org/api/1/origin/save/" 

15SAVE_VISIT_TYPE = "git" 

16SWHID_ORIGIN_SAFE_CHARS = ":/?[]@!$&'()*+,=" 

17 

18 

19def archive_origin(origin_url: str, timeout_seconds: int, poll_interval_seconds: int) -> str: 

20 session = requests.Session() 

21 save_request = _request_save(session, origin_url) 

22 deadline = time.monotonic() + timeout_seconds 

23 current = save_request 

24 while time.monotonic() <= deadline: 

25 task_status = _required_text(current, "save_task_status") 

26 if task_status == "succeeded": 

27 snapshot_swhid = _required_text(current, "snapshot_swhid") 

28 return _swh_url(snapshot_swhid, origin_url) 

29 if task_status == "failed": 

30 msg = f"Software Heritage save task failed for {origin_url}" 

31 raise RuntimeError(msg) 

32 time.sleep(poll_interval_seconds) 

33 current = _poll_save_request(session, current) 

34 msg = f"Software Heritage save task timed out for {origin_url}" 

35 raise TimeoutError(msg) 

36 

37 

38def _request_save(session: requests.Session, origin_url: str) -> Mapping[str, object]: 

39 response = session.post(_save_url(origin_url), headers={"Accept": "application/json"}, timeout=30) 

40 response.raise_for_status() 

41 return cast("Mapping[str, object]", response.json()) 

42 

43 

44def _poll_save_request( 

45 session: requests.Session, 

46 current: Mapping[str, object], 

47) -> Mapping[str, object]: 

48 url = _required_text(current, "request_url") 

49 response = session.get(url, headers={"Accept": "application/json"}, timeout=30) 

50 response.raise_for_status() 

51 data = response.json() 

52 if isinstance(data, list): 

53 return cast("Mapping[str, object]", data[0]) 

54 return cast("Mapping[str, object]", data) 

55 

56 

57def _save_url(origin_url: str) -> str: 

58 return f"{SAVE_API}{SAVE_VISIT_TYPE}/url/{quote(origin_url, safe='')}/" 

59 

60 

61def _swh_url(snapshot_swhid: str, origin_url: str) -> str: 

62 encoded_origin = quote(origin_url, safe=SWHID_ORIGIN_SAFE_CHARS) 

63 return f"https://archive.softwareheritage.org/{snapshot_swhid};origin={encoded_origin}" 

64 

65 

66def _required_text(data: Mapping[str, object], key: str) -> str: 

67 return cast("str", data[key])