|
1 |
| -import re |
2 | 1 | from datetime import datetime, timedelta, timezone
|
3 | 2 | from os import path
|
4 | 3 | from urllib.parse import urlparse
|
@@ -37,22 +36,25 @@ def _request(self, url, **kwargs):
|
37 | 36 | return self.session.get(url, **kwargs)
|
38 | 37 |
|
39 | 38 | urlopen = _request
|
40 |
| - url_regex = r"/dataset/[a-z0-9_\\-]*$" |
41 | 39 |
|
42 | 40 | def detect(self, source, ref=None, extra_args=None):
|
43 | 41 | """Trigger this provider for things that resolve to a CKAN dataset."""
|
44 | 42 | parsed_url = urlparse(source)
|
45 | 43 | if not parsed_url.netloc:
|
46 | 44 | return None
|
47 | 45 |
|
48 |
| - api_url = parsed_url._replace( |
49 |
| - path=re.sub(self.url_regex, "/api/3/action/", parsed_url.path) |
50 |
| - ).geturl() |
| 46 | + url_parts = parsed_url.path.split("/") |
| 47 | + if url_parts[-2] == "dataset": |
| 48 | + self.dataset_id = url_parts[-1] |
| 49 | + else: |
| 50 | + return None |
| 51 | + |
| 52 | + api_url_path = "/api/3/action/" |
| 53 | + api_url = parsed_url._replace(path=api_url_path).geturl() |
51 | 54 |
|
52 | 55 | status_show_url = f"{api_url}status_show"
|
53 | 56 | resp = self.urlopen(status_show_url)
|
54 | 57 | if resp.status_code == 200:
|
55 |
| - self.dataset_id = parsed_url.path.rsplit("/", maxsplit=1)[1] |
56 | 58 | self.version = self._fetch_version(api_url)
|
57 | 59 | return {
|
58 | 60 | "dataset_id": self.dataset_id,
|
|
0 commit comments