mirror of
https://github.com/haiwen/seafile-server.git
synced 2025-04-27 19:15:07 +00:00
110 lines
2.1 KiB
Python
110 lines
2.1 KiB
Python
#coding: UTF-8
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from os.path import abspath, basename, exists, expanduser, join
|
|
from subprocess import PIPE, CalledProcessError, Popen
|
|
|
|
import requests
|
|
import termcolor
|
|
|
|
try:
|
|
from functools import lru_cache
|
|
except ImportError:
|
|
from backports.functools_lru_cache import lru_cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _color(s, color):
|
|
return s if not os.isatty(sys.stdout.fileno()) \
|
|
else termcolor.colored(str(s), color)
|
|
|
|
|
|
def green(s):
|
|
return _color(s, 'green')
|
|
|
|
|
|
def red(s):
|
|
return _color(s, 'red')
|
|
|
|
|
|
def debug(fmt, *a):
|
|
logger.debug(green(fmt), *a)
|
|
|
|
|
|
def info(fmt, *a):
|
|
logger.info(green(fmt), *a)
|
|
|
|
|
|
def warning(fmt, *a):
|
|
logger.warn(red(fmt), *a)
|
|
|
|
|
|
def shell(cmd, inputdata=None, wait=True, **kw):
|
|
info('calling "%s" in %s', cmd, kw.get('cwd', os.getcwd()))
|
|
kw['shell'] = not isinstance(cmd, list)
|
|
kw['stdin'] = PIPE if inputdata else None
|
|
p = Popen(cmd, **kw)
|
|
if inputdata:
|
|
p.communicate(inputdata)
|
|
if wait:
|
|
p.wait()
|
|
if p.returncode:
|
|
raise CalledProcessError(p.returncode, cmd)
|
|
else:
|
|
return p
|
|
|
|
|
|
@contextmanager
|
|
def cd(path):
|
|
olddir = os.getcwd()
|
|
os.chdir(path)
|
|
try:
|
|
yield
|
|
finally:
|
|
os.chdir(olddir)
|
|
|
|
|
|
def chdir(func):
|
|
def wrapped(self, *w, **kw):
|
|
with cd(self.projectdir):
|
|
return func(self, *w, **kw)
|
|
|
|
return wrapped
|
|
|
|
|
|
def setup_logging():
|
|
kw = {
|
|
'format': '[%(asctime)s][%(module)s]: %(message)s',
|
|
'datefmt': '%m/%d/%Y %H:%M:%S',
|
|
'level': logging.DEBUG,
|
|
'stream': sys.stdout,
|
|
}
|
|
|
|
logging.basicConfig(**kw)
|
|
logging.getLogger('requests.packages.urllib3.connectionpool'
|
|
).setLevel(logging.WARNING)
|
|
|
|
|
|
def mkdirs(*paths):
|
|
for path in paths:
|
|
if not exists(path):
|
|
os.mkdir(path)
|
|
|
|
def on_travis():
|
|
return 'TRAVIS_BUILD_NUMBER' in os.environ
|
|
|
|
@contextmanager
|
|
def cd(path):
|
|
path = expanduser(path)
|
|
olddir = os.getcwd()
|
|
os.chdir(path)
|
|
try:
|
|
yield
|
|
finally:
|
|
os.chdir(olddir)
|