apb_extra_utils.github 

  1import functools
  2import json
  3import os
  4import shutil
  5from tempfile import mkdtemp
  6from urllib.error import HTTPError
  7from urllib.request import Request, urlopen
  8
  9from apb_extra_utils.misc import download_and_unzip, remove_content_dir, zip_dir, create_dir
 10
 11PREFIX_FILE_LAST_TAG_REPO = 'last_tag_repo_github_'
 12from pathlib import Path
 13
 14TEXT_EXTS = {
 15    ".py", ".txt", ".md", ".rst", ".json", ".yaml", ".yml",
 16    ".ini", ".cfg", ".toml", ".csv", ".tsv", ".xml", ".html",
 17    ".css", ".js", ".sql", ".bat", ".cmd", ".ps1"
 18}
 19
 20def _is_probably_binary(data: bytes) -> bool:
 21    return b"\x00" in data
 22
 23def convert_tree_to_crlf(root_dir: str):
 24    """
 25    Convert endlines LF to CRLF
 26
 27    Args:
 28        root_dir (str): root directory of tree:
 29
 30    Returns:
 31        None
 32    """
 33    root = Path(root_dir)
 34    for p in root.rglob("*"):
 35        if not p.is_file():
 36            continue
 37        if p.suffix.lower() not in TEXT_EXTS:
 38            continue
 39
 40        raw = p.read_bytes()
 41        if _is_probably_binary(raw):
 42            continue
 43
 44        # decode defensivo
 45        try:
 46            text = raw.decode("utf-8")
 47            encoding = "utf-8"
 48        except UnicodeDecodeError:
 49            try:
 50                text = raw.decode("windows-1252")
 51                encoding = "windows-1252"
 52            except UnicodeDecodeError:
 53                continue
 54
 55        # normaliza primero a LF y luego a CRLF
 56        text = text.replace("\r\n", "\n").replace("\r", "\n")
 57        text = text.replace("\n", "\r\n")
 58
 59        p.write_text(text, encoding=encoding, newline="")
 60
 61
 62def get_api_github(owner, repo, api_request, token=None):
 63    """
 64    GET Request for repository GITHUB via API GITHUB.
 65
 66    See the REST API DOCS here https://docs.github.com/en/rest
 67
 68    Args:
 69        owner (str):
 70        repo (str):
 71        api_request (str):
 72        token (str=None):
 73
 74    Returns:
 75        info_response (dict)
 76    """
 77    request_headers = {
 78        'Accept': 'application/vnd.github.v3+json'
 79    }
 80    if token:
 81        request_headers['Authorization'] = f'token {token}'
 82
 83    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
 84    req = Request(url_github, headers=request_headers, method='GET')
 85    info_response = {}
 86    try:
 87        with urlopen(req) as resp_request:
 88            if resp_request:
 89                info_response = json.load(resp_request)
 90    except HTTPError as exc:
 91        info_response[HTTPError.__name__] = str(exc)
 92
 93    return info_response
 94
 95
 96def post_api_github(owner, repo, api_request, post_data, token=None):
 97    """
 98    POST Request on repository GITHUB via API GITHUB
 99
100    See the REST API DOCS here https://docs.github.com/en/rest
101
102    Args:
103        owner (str):
104        repo (str):
105        api_request (str):
106        post_data (dict):
107        token (str=None):
108
109    Returns:
110        info_response (dict)
111    """
112    request_headers = {
113        'Accept': 'application/vnd.github.v3+json',
114        'Content-Type': 'application/json; charset=utf-8',
115    }
116    if token:
117        request_headers['Authorization'] = f'token {token}'
118
119    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
120
121    post_data_enc = json.dumps(post_data).encode('utf-8')
122
123    req = Request(url_github, headers=request_headers, method='POST')
124    info_response = {}
125    try:
126        with urlopen(req, post_data_enc) as resp_request:
127            if resp_request:
128                info_response = resp_request.__dict__
129    except HTTPError as exc:
130        info_response[HTTPError.__name__] = str(exc)
131
132    return info_response
133
134
135@functools.cache
136def has_changes_in_github(owner, repo, branch, download_to, token=None):
137    """
138    Check if the GitHub repository branch has changes.
139    
140    Args:
141        owner (str): Owner repository Github
142        repo (str): Name repository Github
143        branch (str): Branch repository to check
144        download_to (str): Path to the local repository
145        token (str=None): Github token for private access
146
147    Returns:
148        bool: True if there are changes, False otherwise
149        str: sha_commit of the current branch state
150    """
151    branch = branch.lower()
152    info_branches = get_api_github(owner, repo, 'branches', token)
153    info_branch = next(filter(lambda el: el.get('name', '').lower() == branch,
154                              info_branches), None)
155
156    if not info_branch:
157        return False, None
158
159    sha_commit = info_branch.get('commit').get('sha')
160    expected_name_zip_repo = f'{repo}-{branch}'
161    log_last_tag = os.path.join(download_to, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
162
163    if os.path.exists(log_last_tag):
164        with open(log_last_tag) as fr:
165            last_tag = fr.read()
166            if last_tag and last_tag.strip() == sha_commit.strip():
167                return False, sha_commit
168
169    return True, sha_commit
170
171
172def get_resources_from_repo_github(html_repo, tag, expected_name_zip_repo, path_repo, header=None, force_update=False,
173                                   remove_prev=False, as_zip=False, normalize_eol_win32=None):
174    """
175    
176    Args:
177        html_repo (str):
178        tag (str):
179        expected_name_zip_repo (str):
180        path_repo (str):
181        header (dict=None):
182        force_update (bool=False):
183        remove_prev (bool=False):
184        as_zip (bool=False):
185        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
186            si None convierte solo en Windows; si False no convierte.
187
188    Returns:
189        updated (bool)
190    """
191    updated = False
192    log_last_tag = os.path.join(path_repo, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
193
194    if not force_update:
195        last_tag = None
196        if os.path.exists(log_last_tag):
197            with open(log_last_tag) as fr:
198                last_tag = fr.read()
199
200        if last_tag and last_tag.strip() == tag.strip():
201            return updated
202
203    if not header:
204        header = {}
205    header['Accept'] = 'application/octet-stream'
206
207    dir_temp = mkdtemp()
208    download_and_unzip(html_repo, extract_to=dir_temp, headers=[(k, v) for k, v in header.items()])
209    path_res = os.path.join(dir_temp, expected_name_zip_repo)
210
211    if os.path.exists(path_res):
212        create_dir(path_repo)
213
214        if as_zip:
215            zip_dir(path_res, os.path.join(path_repo, f'{expected_name_zip_repo}.zip'))
216        else:
217            if remove_prev and os.path.exists(path_repo):
218                remove_content_dir(path_repo)
219            shutil.copytree(path_res, path_repo, dirs_exist_ok=True)
220            if normalize_eol_win32 is True or (normalize_eol_win32 is None and os.name == 'nt'):
221                convert_tree_to_crlf(path_repo)
222
223        shutil.rmtree(path_res, ignore_errors=True)
224
225        with open(log_last_tag, "w+") as fw:
226            fw.write(tag)
227
228        updated = True
229
230    return updated
231
232
233def download_release_repo_github(owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False,
234                                 remove_prev=False, normalize_eol_win32=None):
235    """
236    Download release Github repository on the path selected.
237
238    Args:
239        owner (str): Owner repository Github
240        repo (str): Name repository Github
241        download_to (str): Path to download
242        tag_release (str=None): if not informed get 'latest' release
243        token (str=None): Github token for private access
244        force (bool=False): Force update if exists previous sources
245        remove_prev (bool=False): Remove all previous resources
246        as_zip (bool=False): Retorna como ZIP
247        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
248            si None convierte solo en Windows; si False no convierte.
249
250    Returns:
251        tag_name (str)
252    """
253    if not tag_release:
254        info_release = get_api_github(owner, repo, 'releases/latest', token)
255    else:
256        info_release = get_api_github(owner, repo, f'releases/tags/{tag_release}', token)
257
258    tag_name = info_release.get('tag_name')
259    if tag_name:
260        html_release = f'https://github.com/{owner}/{repo}/archive/refs/tags/{tag_name}.zip'
261        header = {}
262        if token:
263            header['Authorization'] = f'token {token}'
264
265        get_resources_from_repo_github(html_release, tag_name, f'{repo}-{tag_name}', download_to, header=header,
266                                       force_update=force, remove_prev=remove_prev, as_zip=as_zip, normalize_eol_win32=normalize_eol_win32)
267
268        return tag_name
269
270
271def download_branch_repo_github(owner, repo, branch, download_to, token=None, force=False, as_zip=False,
272                                remove_prev=False, normalize_eol_win32=None):
273    """
274    Download the branch selected for the Github repo on the path selected
275
276    Args:
277        owner (str): Owner repository Github
278        repo (str): Name repository Github
279        branch (str): Branch repository to download
280        download_to (str): Path to download
281        token (str=None): Github token for private access
282        force (bool=False): Force update if exists previous sources
283        remove_prev (bool=False): Remove all previous resources
284        as_zip (bool=False): Retorna como ZIP
285        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
286            si None convierte solo en Windows; si False no convierte.
287
288    Returns:
289        sha_commit (str), updated (boolean)
290    """
291    has_changes, sha_commit = has_changes_in_github(owner, repo, branch, download_to, token)
292    if not has_changes and not force:
293        return sha_commit, False
294    html_branch = f'https://github.com/{owner}/{repo}/archive/refs/heads/{branch}.zip'
295    header = {}
296    if token:
297        header['Authorization'] = f'token {token}'
298
299    name_zip = f'{repo}-{branch}'
300    updated = get_resources_from_repo_github(html_branch, sha_commit, name_zip, download_to, header=header,
301                                             force_update=force, remove_prev=remove_prev, as_zip=as_zip, normalize_eol_win32=normalize_eol_win32)
302
303    if as_zip:
304        path_zip = os.path.join(download_to, f'{name_zip}.zip')
305        if os.path.exists(path_zip):
306            new_path_zip = os.path.join(download_to, f'{name_zip}-{sha_commit}.zip')
307            if os.path.exists(new_path_zip):
308                os.remove(new_path_zip)
309            os.rename(path_zip, new_path_zip)
310
311    return sha_commit, updated
312
313
314if __name__ == '__main__':
315    import fire, sys
316
317    sys.exit(fire.Fire(
318        {
319            get_api_github.__name__: get_api_github,
320            post_api_github.__name__: post_api_github,
321            download_release_repo_github.__name__: download_release_repo_github,
322            download_branch_repo_github.__name__: download_branch_repo_github
323        }
324    ))
PREFIX_FILE_LAST_TAG_REPO = 'last_tag_repo_github_'
TEXT_EXTS = {'.cfg', '.py', '.toml', '.bat', '.yaml', '.txt', '.json', '.tsv', '.rst', '.yml', '.ini', '.xml', '.sql', '.md', '.ps1', '.cmd', '.js', '.html', '.css', '.csv'}
def convert_tree_to_crlf(root_dir: str): 
24def convert_tree_to_crlf(root_dir: str):
25    """
26    Convert endlines LF to CRLF
27
28    Args:
29        root_dir (str): root directory of tree:
30
31    Returns:
32        None
33    """
34    root = Path(root_dir)
35    for p in root.rglob("*"):
36        if not p.is_file():
37            continue
38        if p.suffix.lower() not in TEXT_EXTS:
39            continue
40
41        raw = p.read_bytes()
42        if _is_probably_binary(raw):
43            continue
44
45        # decode defensivo
46        try:
47            text = raw.decode("utf-8")
48            encoding = "utf-8"
49        except UnicodeDecodeError:
50            try:
51                text = raw.decode("windows-1252")
52                encoding = "windows-1252"
53            except UnicodeDecodeError:
54                continue
55
56        # normaliza primero a LF y luego a CRLF
57        text = text.replace("\r\n", "\n").replace("\r", "\n")
58        text = text.replace("\n", "\r\n")
59
60        p.write_text(text, encoding=encoding, newline="")

Convert endlines LF to CRLF

Arguments:
  • root_dir (str): root directory of tree:
Returns:

None

def get_api_github(owner, repo, api_request, token=None): 
63def get_api_github(owner, repo, api_request, token=None):
64    """
65    GET Request for repository GITHUB via API GITHUB.
66
67    See the REST API DOCS here https://docs.github.com/en/rest
68
69    Args:
70        owner (str):
71        repo (str):
72        api_request (str):
73        token (str=None):
74
75    Returns:
76        info_response (dict)
77    """
78    request_headers = {
79        'Accept': 'application/vnd.github.v3+json'
80    }
81    if token:
82        request_headers['Authorization'] = f'token {token}'
83
84    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
85    req = Request(url_github, headers=request_headers, method='GET')
86    info_response = {}
87    try:
88        with urlopen(req) as resp_request:
89            if resp_request:
90                info_response = json.load(resp_request)
91    except HTTPError as exc:
92        info_response[HTTPError.__name__] = str(exc)
93
94    return info_response

GET Request for repository GITHUB via API GITHUB.

See the REST API DOCS here apb_extra_utils.github.com/en/rest">https://docsapb_extra_utils.github.com/en/rest

Arguments:
  • owner (str):
  • repo (str):
  • api_request (str):
  • token (str=None):
Returns:

info_response (dict)

def post_api_github(owner, repo, api_request, post_data, token=None): 
 97def post_api_github(owner, repo, api_request, post_data, token=None):
 98    """
 99    POST Request on repository GITHUB via API GITHUB
100
101    See the REST API DOCS here https://docs.github.com/en/rest
102
103    Args:
104        owner (str):
105        repo (str):
106        api_request (str):
107        post_data (dict):
108        token (str=None):
109
110    Returns:
111        info_response (dict)
112    """
113    request_headers = {
114        'Accept': 'application/vnd.github.v3+json',
115        'Content-Type': 'application/json; charset=utf-8',
116    }
117    if token:
118        request_headers['Authorization'] = f'token {token}'
119
120    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
121
122    post_data_enc = json.dumps(post_data).encode('utf-8')
123
124    req = Request(url_github, headers=request_headers, method='POST')
125    info_response = {}
126    try:
127        with urlopen(req, post_data_enc) as resp_request:
128            if resp_request:
129                info_response = resp_request.__dict__
130    except HTTPError as exc:
131        info_response[HTTPError.__name__] = str(exc)
132
133    return info_response

POST Request on repository GITHUB via API GITHUB

See the REST API DOCS here apb_extra_utils.github.com/en/rest">https://docsapb_extra_utils.github.com/en/rest

Arguments:
  • owner (str):
  • repo (str):
  • api_request (str):
  • post_data (dict):
  • token (str=None):
Returns:

info_response (dict)

@functools.cache
def has_changes_in_github(owner, repo, branch, download_to, token=None): 
136@functools.cache
137def has_changes_in_github(owner, repo, branch, download_to, token=None):
138    """
139    Check if the GitHub repository branch has changes.
140    
141    Args:
142        owner (str): Owner repository Github
143        repo (str): Name repository Github
144        branch (str): Branch repository to check
145        download_to (str): Path to the local repository
146        token (str=None): Github token for private access
147
148    Returns:
149        bool: True if there are changes, False otherwise
150        str: sha_commit of the current branch state
151    """
152    branch = branch.lower()
153    info_branches = get_api_github(owner, repo, 'branches', token)
154    info_branch = next(filter(lambda el: el.get('name', '').lower() == branch,
155                              info_branches), None)
156
157    if not info_branch:
158        return False, None
159
160    sha_commit = info_branch.get('commit').get('sha')
161    expected_name_zip_repo = f'{repo}-{branch}'
162    log_last_tag = os.path.join(download_to, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
163
164    if os.path.exists(log_last_tag):
165        with open(log_last_tag) as fr:
166            last_tag = fr.read()
167            if last_tag and last_tag.strip() == sha_commit.strip():
168                return False, sha_commit
169
170    return True, sha_commit

Check if the GitHub repository branch has changes.

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • branch (str): Branch repository to check
  • download_to (str): Path to the local repository
  • token (str=None): Github token for private access
Returns:

bool: True if there are changes, False otherwise str: sha_commit of the current branch state

def get_resources_from_repo_github( html_repo, tag, expected_name_zip_repo, path_repo, header=None, force_update=False, remove_prev=False, as_zip=False, normalize_eol_win32=None): 
173def get_resources_from_repo_github(html_repo, tag, expected_name_zip_repo, path_repo, header=None, force_update=False,
174                                   remove_prev=False, as_zip=False, normalize_eol_win32=None):
175    """
176    
177    Args:
178        html_repo (str):
179        tag (str):
180        expected_name_zip_repo (str):
181        path_repo (str):
182        header (dict=None):
183        force_update (bool=False):
184        remove_prev (bool=False):
185        as_zip (bool=False):
186        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
187            si None convierte solo en Windows; si False no convierte.
188
189    Returns:
190        updated (bool)
191    """
192    updated = False
193    log_last_tag = os.path.join(path_repo, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
194
195    if not force_update:
196        last_tag = None
197        if os.path.exists(log_last_tag):
198            with open(log_last_tag) as fr:
199                last_tag = fr.read()
200
201        if last_tag and last_tag.strip() == tag.strip():
202            return updated
203
204    if not header:
205        header = {}
206    header['Accept'] = 'application/octet-stream'
207
208    dir_temp = mkdtemp()
209    download_and_unzip(html_repo, extract_to=dir_temp, headers=[(k, v) for k, v in header.items()])
210    path_res = os.path.join(dir_temp, expected_name_zip_repo)
211
212    if os.path.exists(path_res):
213        create_dir(path_repo)
214
215        if as_zip:
216            zip_dir(path_res, os.path.join(path_repo, f'{expected_name_zip_repo}.zip'))
217        else:
218            if remove_prev and os.path.exists(path_repo):
219                remove_content_dir(path_repo)
220            shutil.copytree(path_res, path_repo, dirs_exist_ok=True)
221            if normalize_eol_win32 is True or (normalize_eol_win32 is None and os.name == 'nt'):
222                convert_tree_to_crlf(path_repo)
223
224        shutil.rmtree(path_res, ignore_errors=True)
225
226        with open(log_last_tag, "w+") as fw:
227            fw.write(tag)
228
229        updated = True
230
231    return updated
Arguments:
  • html_repo (str):
  • tag (str):
  • expected_name_zip_repo (str):
  • path_repo (str):
  • header (dict=None):
  • force_update (bool=False):
  • remove_prev (bool=False):
  • as_zip (bool=False):
  • normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión; si None convierte solo en Windows; si False no convierte.
Returns:

updated (bool)

def download_release_repo_github( owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False, remove_prev=False, normalize_eol_win32=None): 
234def download_release_repo_github(owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False,
235                                 remove_prev=False, normalize_eol_win32=None):
236    """
237    Download release Github repository on the path selected.
238
239    Args:
240        owner (str): Owner repository Github
241        repo (str): Name repository Github
242        download_to (str): Path to download
243        tag_release (str=None): if not informed get 'latest' release
244        token (str=None): Github token for private access
245        force (bool=False): Force update if exists previous sources
246        remove_prev (bool=False): Remove all previous resources
247        as_zip (bool=False): Retorna como ZIP
248        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
249            si None convierte solo en Windows; si False no convierte.
250
251    Returns:
252        tag_name (str)
253    """
254    if not tag_release:
255        info_release = get_api_github(owner, repo, 'releases/latest', token)
256    else:
257        info_release = get_api_github(owner, repo, f'releases/tags/{tag_release}', token)
258
259    tag_name = info_release.get('tag_name')
260    if tag_name:
261        html_release = f'https://github.com/{owner}/{repo}/archive/refs/tags/{tag_name}.zip'
262        header = {}
263        if token:
264            header['Authorization'] = f'token {token}'
265
266        get_resources_from_repo_github(html_release, tag_name, f'{repo}-{tag_name}', download_to, header=header,
267                                       force_update=force, remove_prev=remove_prev, as_zip=as_zip, normalize_eol_win32=normalize_eol_win32)
268
269        return tag_name

Download release Github repository on the path selected.

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • download_to (str): Path to download
  • tag_release (str=None): if not informed get 'latest' release
  • token (str=None): Github token for private access
  • force (bool=False): Force update if exists previous sources
  • remove_prev (bool=False): Remove all previous resources
  • as_zip (bool=False): Retorna como ZIP
  • normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión; si None convierte solo en Windows; si False no convierte.
Returns:

tag_name (str)

def download_branch_repo_github( owner, repo, branch, download_to, token=None, force=False, as_zip=False, remove_prev=False, normalize_eol_win32=None): 
272def download_branch_repo_github(owner, repo, branch, download_to, token=None, force=False, as_zip=False,
273                                remove_prev=False, normalize_eol_win32=None):
274    """
275    Download the branch selected for the Github repo on the path selected
276
277    Args:
278        owner (str): Owner repository Github
279        repo (str): Name repository Github
280        branch (str): Branch repository to download
281        download_to (str): Path to download
282        token (str=None): Github token for private access
283        force (bool=False): Force update if exists previous sources
284        remove_prev (bool=False): Remove all previous resources
285        as_zip (bool=False): Retorna como ZIP
286        normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión;
287            si None convierte solo en Windows; si False no convierte.
288
289    Returns:
290        sha_commit (str), updated (boolean)
291    """
292    has_changes, sha_commit = has_changes_in_github(owner, repo, branch, download_to, token)
293    if not has_changes and not force:
294        return sha_commit, False
295    html_branch = f'https://github.com/{owner}/{repo}/archive/refs/heads/{branch}.zip'
296    header = {}
297    if token:
298        header['Authorization'] = f'token {token}'
299
300    name_zip = f'{repo}-{branch}'
301    updated = get_resources_from_repo_github(html_branch, sha_commit, name_zip, download_to, header=header,
302                                             force_update=force, remove_prev=remove_prev, as_zip=as_zip, normalize_eol_win32=normalize_eol_win32)
303
304    if as_zip:
305        path_zip = os.path.join(download_to, f'{name_zip}.zip')
306        if os.path.exists(path_zip):
307            new_path_zip = os.path.join(download_to, f'{name_zip}-{sha_commit}.zip')
308            if os.path.exists(new_path_zip):
309                os.remove(new_path_zip)
310            os.rename(path_zip, new_path_zip)
311
312    return sha_commit, updated

Download the branch selected for the Github repo on the path selected

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • branch (str): Branch repository to download
  • download_to (str): Path to download
  • token (str=None): Github token for private access
  • force (bool=False): Force update if exists previous sources
  • remove_prev (bool=False): Remove all previous resources
  • as_zip (bool=False): Retorna como ZIP
  • normalize_eol_win32 (bool=None): Normaliza endline para CRLF. Si True fuerza la conversión; si None convierte solo en Windows; si False no convierte.
Returns:

sha_commit (str), updated (boolean)