Python utils 模块，random_string() 实例源码

def proxy_file_downloader(request):
    if not is_admin_or_root(request.user):
        raise PermissionDenied

    def download_file(url):
        local_filename = url.split('/')[-1]
        if local_filename == '':
            local_filename = random_string()
        r = requests.get(url, stream=True, timeout=30)
        with open(path.join(settings.UPLOAD_DIR, local_filename), 'wb') as f:
            for chunk in r.iter_content(chunk_size=1024):
                if chunk:
                    f.write(chunk)

    if request.method == 'POST':
        try:
            url = request.POST['url']
            Thread(target=download_file, args=(url,)).start()
        except Exception as e:
            raise PermissionDenied(repr(e))
    return redirect(reverse('filemanager'))

def create_session(self):
        session_id = random_string(20)
        session = HttpSession(self.dispatcher, session_id)
        return session_id

def prepape_auth_headers(session, include_cauth=False):
    """
    This function prepapes headers with CSRF/CAUTH tokens that can
    be used in POST requests such as login/get_quiz.
    @param session: Requests session.
    @type session: requests.Session
    @param include_cauth: Flag that indicates whethe CAUTH cookies should be
        included as well.
    @type include_cauth: bool
    @return: Dictionary of headers.
    @rtype: dict
    """

    # csrftoken is simply a 20 char random string
    csrftoken = random_string(20)

    # now make a call to the authenticator url
    csrf2cookie = 'csrf2_token_%s' % random_string(8)
    csrf2token = random_string(24)
    cookie = "csrftoken=%s; %s=%s" % (csrftoken, csrf2cookie, csrf2token)

    if include_cauth:
        CAUTH = session.cookies.get('CAUTH')
        cookie = "CAUTH=%s; %s" % (CAUTH, cookie)

    logging.debug('Forging cookie header: %s.', cookie)
    headers = {
        'Cookie': cookie,
        'X-CSRFToken': csrftoken,
        'X-CSRF2-Cookie': csrf2cookie,
        'X-CSRF2-Token': csrf2token
    }

    return headers

def create_session(self):
        session_id = random_string(20)
        session = HttpSession(self.dispatcher, session_id)
        return session_id

def __init__(self, contact, text, author=None, time=None, read=False, id=None):
        # system messages have author=None
        self.id, self.text, self.author, self.contact = utils.random_string(10) if not id else id, text, author, contact
        self.time, self.read = time if time is not None else datetime.now(), read

def get(self, request, cid):
        type = request.GET.get('t')
        if type and 'all' in type and self.privileged:
            submissions = self.contest.submission_set
        else:
            submissions = self.contest.submission_set.filter(author=request.user)
        if type and 'accepted' in type:
            submissions = submissions.filter(status=SubmissionStatus.ACCEPTED)
        submissions = submissions.select_related("author")
        self.contest.add_contest_problem_to_submissions(submissions)
        participants = dict(self.contest.contestparticipant_set.values_list('user_id', 'comment'))

        file_path = path.join(settings.GENERATE_DIR, random_string())
        lang_ext_dict = dict(LANG_EXT)
        with zipfile.ZipFile(file_path, "w", zipfile.ZIP_DEFLATED) as zip:
            zip.writestr('/contest.nfo', '')
            for submission in submissions:
                user = submission.author.username
                if participants[submission.author_id]:
                    user = participants[submission.author_id]
                user = self.__class__.slugify_filename(user)
                if getattr(submission, 'contest_problem') and submission.contest_problem:
                    zip.writestr("/%s_%s/%s_#%d_%s.%s" % (user, submission.author_id, submission.contest_problem.identifier,
                                                          submission.pk, submission.get_status_display().replace(' ', '_'),
                                                          lang_ext_dict.get(submission.lang, 'txt')),
                                 submission.code)
        return respond_generate_file(request, file_path, "ContestCode - %s.zip" % self.contest.title)

def _prepare_judge_json_data(code, lang, max_time, max_memory, run_until_complete, cases, checker, interactor):
    all_params = locals().copy()
    if not interactor:
        all_params.pop('interactor')
    all_params['max_time'] /= 1000
    all_params['fingerprint'] = random_string()
    return all_params

def post(self, request, *args, **kwargs):
        files = request.FILES.getlist("files[]")
        for file in files:
            save_uploaded_file_to(file, path.join(settings.UPLOAD_DIR, str(self.problem.pk)),
                                  filename=path.splitext(file.name)[0] + '.' + random_string(16),
                                  keep_extension=True)
        return HttpResponse()

def post(self, request, *args, **kwargs):
        if request.POST['type'] == 'manual':
            input = request.POST['input']
            output = request.POST['output']
            well_form = request.POST.get("wellForm") == "on"
            if well_form:
                input, output = well_form_text(input), well_form_text(output)
            if not input:
                raise ValueError('Input file cannot be empty')
            save_case(self.session, input.encode(), output.encode(), well_form=well_form)
        elif request.POST['type'] == 'upload':
            file = request.FILES['file']
            file_directory = '/tmp'
            file_path = save_uploaded_file_to(file, file_directory, filename=random_string(), keep_extension=True)
            process_uploaded_case(self.session, file_path)
            remove(file_path)
        elif request.POST['type'] == 'generate':
            generator = request.POST['generator']
            raw_param = request.POST['param']
            generate_input('Generate cases', self.session, generator, raw_param)
        elif request.POST['type'] == 'stress':
            generator = request.POST['generator']
            raw_param = request.POST['param']
            submission = request.POST['submission']
            time = int(request.POST['time']) * 60
            if time < 60 or time > 300:
                raise ValueError('Time not in range')
            stress('Stress test', self.session, generator, submission, raw_param, time)
        return HttpResponse()

def init_session(problem, user):
    """
    Init a session
    :type problem: Problem
    :type user: User
    :return: session
    """
    fingerprint = random_string()
    session = EditSession.objects.create(problem=problem, user=user, fingerprint=fingerprint,
                                         last_synchronize=datetime.now())
    rmtree(path.join(settings.REPO_DIR, fingerprint), ignore_errors=True)
    makedirs(path.join(settings.REPO_DIR, fingerprint))
    pull_session(session)
    return session

def process_uploaded_case(session, file_path):
    if re.search(r'\.zip$', file_path, re.IGNORECASE):
        # this is a zip file
        tmp_directory = '/tmp/' + random_string()
        with zipfile.ZipFile(file_path) as myZip:
            myZip.extractall(path=tmp_directory)
        for inf, ouf in sort_data_list_from_directory(tmp_directory):
            with open(path.join(tmp_directory, inf), 'rb') as ins, open(path.join(tmp_directory, ouf), 'rb') as ous:
                save_case(session, ins.read(), ous.read())
        rmtree(tmp_directory)
    else:
        with open(file_path, 'rb') as file:
            save_case(session, file.read(), b'')