GodsDevProject commited on
Commit
0c4faef
·
verified ·
1 Parent(s): 2568db7

Update ingest/generic_public_foia.py

Browse files
Files changed (1) hide show
  1. ingest/generic_public_foia.py +4 -41
ingest/generic_public_foia.py CHANGED
@@ -1,5 +1,3 @@
1
- # ingest/generic_public_foia.py
2
-
3
  import abc
4
  import asyncio
5
  from typing import List, Dict
@@ -8,70 +6,35 @@ from typing import List, Dict
8
  class GenericFOIAAdapter(abc.ABC):
9
  """
10
  Base adapter for all public FOIA Electronic Reading Rooms.
11
-
12
- This class is SAFE for Hugging Face Spaces:
13
- - Public-only access
14
- - Rate-limited
15
- - No authentication
16
- - Explicit stub/live disclosure
17
  """
18
 
19
- # Human-readable source name (shown in UI)
20
  source_name: str = "UNKNOWN"
21
-
22
- # Base public reading room URL
23
  base_url: str = ""
24
-
25
- # Either "stub" or "live"
26
- source_type: str = "stub"
27
 
28
  def __init__(self, rate_limit_seconds: float = 1.0):
29
  self.rate_limit_seconds = rate_limit_seconds
30
  self._last_call = 0.0
31
 
32
  async def _rate_limit(self):
33
- """
34
- Enforce per-adapter rate limiting (HF-safe).
35
- """
36
  now = asyncio.get_event_loop().time()
37
  delta = now - self._last_call
38
  if delta < self.rate_limit_seconds:
39
  await asyncio.sleep(self.rate_limit_seconds - delta)
40
  self._last_call = asyncio.get_event_loop().time()
41
 
42
- @property
43
- def is_live(self) -> bool:
44
- """
45
- Convenience flag used by UI and filters.
46
- """
47
- return self.source_type == "live"
48
-
49
  async def _stub_result(self, query: str) -> List[Dict]:
50
- """
51
- Default placeholder result for stub adapters.
52
- """
53
  await self._rate_limit()
54
  return [{
55
  "source": self.source_name,
56
  "title": f"{self.source_name} FOIA result for '{query}'",
57
  "url": self.base_url,
58
  "snippet": "Public FOIA reading room placeholder result (stub).",
59
- "live": False
 
60
  }]
61
 
62
  @abc.abstractmethod
63
  async def search(self, query: str) -> List[Dict]:
64
- """
65
- Perform a public FOIA search.
66
-
67
- Must return a list of dicts with:
68
- - source (str)
69
- - title (str)
70
- - url (str)
71
- - snippet (str)
72
- - live (bool)
73
-
74
- Stub adapters should normally call self._stub_result(query).
75
- Live adapters should set live=True.
76
- """
77
  raise NotImplementedError
 
 
 
1
  import abc
2
  import asyncio
3
  from typing import List, Dict
 
6
  class GenericFOIAAdapter(abc.ABC):
7
  """
8
  Base adapter for all public FOIA Electronic Reading Rooms.
9
+ HF-safe: public-only, rate-limited, explicit stub/live labeling.
 
 
 
 
 
10
  """
11
 
 
12
  source_name: str = "UNKNOWN"
 
 
13
  base_url: str = ""
14
+ source_type: str = "stub" # stub | live
 
 
15
 
16
  def __init__(self, rate_limit_seconds: float = 1.0):
17
  self.rate_limit_seconds = rate_limit_seconds
18
  self._last_call = 0.0
19
 
20
  async def _rate_limit(self):
 
 
 
21
  now = asyncio.get_event_loop().time()
22
  delta = now - self._last_call
23
  if delta < self.rate_limit_seconds:
24
  await asyncio.sleep(self.rate_limit_seconds - delta)
25
  self._last_call = asyncio.get_event_loop().time()
26
 
 
 
 
 
 
 
 
27
  async def _stub_result(self, query: str) -> List[Dict]:
 
 
 
28
  await self._rate_limit()
29
  return [{
30
  "source": self.source_name,
31
  "title": f"{self.source_name} FOIA result for '{query}'",
32
  "url": self.base_url,
33
  "snippet": "Public FOIA reading room placeholder result (stub).",
34
+ "live": False,
35
+ "extended": False
36
  }]
37
 
38
  @abc.abstractmethod
39
  async def search(self, query: str) -> List[Dict]:
 
 
 
 
 
 
 
 
 
 
 
 
 
40
  raise NotImplementedError